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

Merged in feature/black-and-isort (pull request #717)

Browse Source 
Feature/black and isort

* black and isort constants

* black and isort codes

* black and isort crosswalks

* black and isort investment

* black and isort prompts

* isort testbed

* isort tracking

* black and isort utils

* poetry and black tests


Approved-by: Katon Minhas
This commit is contained in:
Alex Galarce
2025-09-30 20:56:02 +00:00
parent ae4eda6f25
commit 674add9a3e
43 changed files with 841 additions and 515 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fieldExtraction/constants/constants.py
+53 -23
View File
@@ -13,33 +13,55 @@ class Constants:
#################################### Dynamic and Exhibit-Level ####################################
# Dynamic Primary
self.CROSSWALK_LOB = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_lob.json")
self.CROSSWALK_LOB = CrosswalkBuilder().from_json(
path="constants/mappings/crosswalk_lob.json"
)
self.VALID_LOBS = list(self.CROSSWALK_LOB.mapping.keys())
self.CROSSWALK_PROGRAM = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_program.json")
self.CROSSWALK_PROGRAM = CrosswalkBuilder().from_json(
path="constants/mappings/crosswalk_program.json"
)
self.VALID_PROGRAMS = list(self.CROSSWALK_PROGRAM.mapping.keys())
self.CROSSWALK_NETWORK = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_network.json")
self.CROSSWALK_NETWORK = CrosswalkBuilder().from_json(
path="constants/mappings/crosswalk_network.json"
)
self.VALID_NETWORKS = list(self.CROSSWALK_NETWORK.mapping.keys())
self.CROSSWALK_PRODUCT = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_product_lob.json")
self.CROSSWALK_PRODUCT = CrosswalkBuilder().from_json(
path="constants/mappings/crosswalk_product_lob.json"
)
self.VALID_PRODUCTS = list(self.CROSSWALK_PRODUCT.mapping.keys())
# Dynamic Codes
self.CROSSWALK_PROV_SPECIALTY_CD = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_provider_specialty.json")
self.VALID_PROV_SPECIALTY = list(self.CROSSWALK_PROV_SPECIALTY_CD.mapping.keys())
self.CROSSWALK_BILL_TYPE_CD = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_bill_type.json")
self.CROSSWALK_PROV_SPECIALTY_CD = CrosswalkBuilder().from_json(
path="constants/mappings/crosswalk_provider_specialty.json"
)
self.VALID_PROV_SPECIALTY = list(
self.CROSSWALK_PROV_SPECIALTY_CD.mapping.keys()
)
self.CROSSWALK_BILL_TYPE_CD = CrosswalkBuilder().from_json(
path="constants/mappings/crosswalk_bill_type.json"
)
self.VALID_BILL_TYPE = list(set(self.CROSSWALK_BILL_TYPE_CD.mapping.values()))
self.CROSSWALK_PLACE_OF_SERVICE_CD = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_place_of_service.json")
self.VALID_PLACE_OF_SERVICE = list(self.CROSSWALK_PLACE_OF_SERVICE_CD.mapping.keys())
self.CROSSWALK_CLAIM_TYPE_CD =CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_claim_type.json")
self.CROSSWALK_PLACE_OF_SERVICE_CD = CrosswalkBuilder().from_json(
path="constants/mappings/crosswalk_place_of_service.json"
)
self.VALID_PLACE_OF_SERVICE = list(
self.CROSSWALK_PLACE_OF_SERVICE_CD.mapping.keys()
)
self.CROSSWALK_CLAIM_TYPE_CD = CrosswalkBuilder().from_json(
path="constants/mappings/crosswalk_claim_type.json"
)
self.VALID_CLAIM_TYPE = list(set(self.CROSSWALK_BILL_TYPE_CD.mapping.keys()))
# Provider Type
self.LIST_VALID_PROV_TYPE = ListBuilder().from_json("constants/lists/valid_prov_type.json").list
self.LIST_VALID_PROV_TYPE = (
ListBuilder().from_json("constants/lists/valid_prov_type.json").list
)
#################################### Code Breakout ####################################
self.SYNONYM_MAP = (
@@ -102,12 +124,20 @@ class Constants:
)
self.GROUPER_APR_DRG_MAPPING = (
CrosswalkBuilder()
.from_excel("constants/mapping_csvs/grouper_cd/drg_mapping_apr-drg.csv", "Code", "Description")
.from_excel(
"constants/mapping_csvs/grouper_cd/drg_mapping_apr-drg.csv",
"Code",
"Description",
)
.mapping
)
self.GROUPER_MS_DRG_MAPPING = (
CrosswalkBuilder()
.from_excel("constants/mapping_csvs/grouper_cd/drg_mapping_ms-drg.csv", "Code", "Description")
.from_excel(
"constants/mapping_csvs/grouper_cd/drg_mapping_ms-drg.csv",
"Code",
"Description",
)
.mapping
)
@@ -250,18 +280,18 @@ class Constants:
def get_constant(self, attr_name: str):
"""
Returns the value of a constant attribute by its string name.
Args:
attr_name (str): The name of the constant attribute to retrieve.
Returns:
The value of the requested attribute, or None if the attribute doesn't exist.
Example:
constants.get_constant("VALID_BILL_TYPE") will return the VALID_BILL_TYPE list.
"""
if not isinstance(attr_name, str):
return None
return None
if hasattr(self, attr_name):
return getattr(self, attr_name)
return None
fieldExtraction/constants/investment_columns.py
+1 -1
View File
@@ -169,5 +169,5 @@ COLUMN_ORDER = [
"PREMIUM_END_DT",
"SEQUESTRATION_TERM",
"SEQUESTRATION_START_DT",
"SEQUESTRATION_END_DT"
"SEQUESTRATION_END_DT",
]
fieldExtraction/constants/regex_patterns.py
+3 -1
View File
@@ -4,7 +4,9 @@ BACKTICK_PATTERN = r"`([^`]+)`"
TRIPLE_BACKTICK_PATTERN = r"```([^`]+)```"
# TIN and NPI Regex Patterns
TIN_PATTERN = r"\b\d{9}\b|\b\d{2}[-.]\d{7}\b|\b\d{3}[-.]\d{2}[-.]\d{4}\b|\b\d{2}-\d{4}-\d{3}\b"
TIN_PATTERN = (
r"\b\d{9}\b|\b\d{2}[-.]\d{7}\b|\b\d{3}[-.]\d{2}[-.]\d{4}\b|\b\d{2}-\d{4}-\d{3}\b"
)
NPI_PATTERN = r"\b\d{10}\b"
# OCR-Correctable Patterns
fieldExtraction/src/codes/code_funcs.py
+22 -13
View File
@@ -2,7 +2,6 @@ import os
import re
import pandas as pd
import src.config as config
import src.prompts.prompt_templates as prompt_templates
import src.utils.llm_utils as llm_utils
@@ -11,6 +10,7 @@ from constants.constants import Constants
from constants.delimiters import Delimiter
from src.prompts.fieldset import FieldSet
def clean_service(service, constants: Constants) -> str:
"""
Cleans the service string by removing unnecessary terms and formatting it for further processing.
@@ -448,11 +448,21 @@ def fill_grouper_cd_desc(answer_dict, constants: Constants):
grouper_cd = eval(grouper_cd) if "[" in grouper_cd else grouper_cd
for code in grouper_cd:
code_without_severity = int(code.split("-")[0]) # Remove severity level if present
if grouper_type == "APR-DRG" and (code_without_severity in constants.GROUPER_APR_DRG_MAPPING):
grouper_descs.append(constants.GROUPER_APR_DRG_MAPPING[code_without_severity])
elif grouper_type == "MS-DRG" and (code_without_severity in constants.GROUPER_MS_DRG_MAPPING):
grouper_descs.append(constants.GROUPER_MS_DRG_MAPPING[code_without_severity])
code_without_severity = int(
code.split("-")[0]
) # Remove severity level if present
if grouper_type == "APR-DRG" and (
code_without_severity in constants.GROUPER_APR_DRG_MAPPING
):
grouper_descs.append(
constants.GROUPER_APR_DRG_MAPPING[code_without_severity]
)
elif grouper_type == "MS-DRG" and (
code_without_severity in constants.GROUPER_MS_DRG_MAPPING
):
grouper_descs.append(
constants.GROUPER_MS_DRG_MAPPING[code_without_severity]
)
if grouper_descs:
answer_dict["GROUPER_CD_DESC"] = "|".join(list(set(grouper_descs)))
@@ -656,7 +666,8 @@ def code_breakout(merged_results: pd.DataFrame, constants: Constants):
# Fill in GROUPER_CD_DESC based on GROUPER_TYPE and GROUPER_CD
answer_dicts_with_code = [
fill_grouper_cd_desc(answer_dict, constants) for answer_dict in answer_dicts_with_code
fill_grouper_cd_desc(answer_dict, constants)
for answer_dict in answer_dicts_with_code
]
return pd.DataFrame(answer_dicts_with_code)
@@ -679,15 +690,14 @@ def grouper_breakout(results_with_code: pd.DataFrame):
file_path=config.FIELD_JSON_PATH, field_type="grouper_breakout"
).print_prompt_dict()
for answer_dict in answer_dicts_with_code:
if (
string_utils.is_empty(answer_dict.get("GROUPER_TYPE"))
and not string_utils.is_empty(answer_dict.get("GROUPER_CD"))
):
if string_utils.is_empty(
answer_dict.get("GROUPER_TYPE")
) and not string_utils.is_empty(answer_dict.get("GROUPER_CD")):
# run groper breakout prompt
grouper_breakout_prompt = prompt_templates.GROUPER_BREAKOUT(
answer_dict.get("SERVICE_TERM", ""),
answer_dict.get("REIMB_TERM", ""),
GROUPER_QUESTIONS
GROUPER_QUESTIONS,
)
claude_answer_raw = llm_utils.invoke_claude(
grouper_breakout_prompt, "sonnet_latest", ""
@@ -701,4 +711,3 @@ def grouper_breakout(results_with_code: pd.DataFrame):
final_answer_dicts.append(answer_dict)
return pd.DataFrame(final_answer_dicts)
fieldExtraction/src/codes/main.py
+1 -2
View File
@@ -1,11 +1,10 @@
import concurrent.futures
from sentence_transformers import SentenceTransformer
import src.codes.code_funcs as code_funcs
import src.config as config
import src.utils.io_utils as io_utils
from constants.constants import Constants
from sentence_transformers import SentenceTransformer
def main():
fieldExtraction/src/crosswalk/crosswalk_builder.py
+1 -2
View File
@@ -2,7 +2,6 @@ import csv
import json
import pandas as pd
from src.config import CLIENT, STATE
@@ -170,4 +169,4 @@ class CrosswalkBuilder:
return {target: "|".join(sources) for target, sources in reverse_dict.items()}
def map_value(self, key_value):
return self.mapping.get(key_value)
return self.mapping.get(key_value)
fieldExtraction/src/investment/aarete_derived.py
-1
View File
@@ -3,7 +3,6 @@ import os
from collections import defaultdict
import pandas as pd
import src.config as config
import src.utils.string_utils as string_utils
from constants.constants import Constants
fieldExtraction/src/investment/dynamic_funcs.py
+4 -3
View File
@@ -1,11 +1,12 @@
import logging
import src.investment.prompt_calls as prompt_calls
import src.utils.llm_utils as llm_utils
import src.utils.string_utils as string_utils
from constants.constants import Constants
from src import config
from src.prompts import prompt_templates
from src.prompts.fieldset import Field, FieldSet
from constants.constants import Constants
import logging
import src.investment.prompt_calls as prompt_calls
def add_full_context_field(one_to_one_fields, field_to_add, answer_dicts):
fieldExtraction/src/investment/lesser_of_funcs.py
+3 -2
View File
@@ -1,7 +1,8 @@
import re
import logging
from src.utils import string_utils
import re
import src.investment.prompt_calls as prompt_calls
from src.utils import string_utils
def has_explicit_lesser_of_carveout(reimb_term: str) -> bool:
fieldExtraction/src/investment/one_to_one_funcs.py
+6 -11
View File
@@ -2,26 +2,21 @@ import logging
import os
import pandas as pd
from rapidfuzz import fuzz
import src.investment.postprocessing_funcs as postprocessing_funcs
import src.investment.prompt_calls as prompt_calls
import src.prompts.prompt_templates as prompt_templates
import src.utils.llm_utils as llm_utils
import src.utils.string_utils as string_utils
from constants.constants import Constants
from constants.delimiters import Delimiter
from rapidfuzz import fuzz
from src import config
from src.investment import postprocessing_funcs
import src.investment.prompt_calls as prompt_calls
from constants.constants import Constants
from src.investment.smart_chunking_funcs import (
field_context,
group_fields,
stitch_chunks,
)
from src.investment.smart_chunking_funcs import (field_context, group_fields,
stitch_chunks)
from src.investment.vision_funcs import get_image_array_based_answer
from src.prompts.prompt_templates import METHODOLOGY_BREAKOUT
from src.prompts.fieldset import Field, FieldSet
from src.prompts.prompt_templates import METHODOLOGY_BREAKOUT
def extract_global_lesser_of(contract_text: str, filename: str) -> dict:
fieldExtraction/src/investment/postprocessing_funcs.py
+1 -3
View File
@@ -7,15 +7,13 @@ from datetime import datetime
from functools import cache
import pandas as pd
import src.prompts.prompt_templates as prompt_templates
import src.utils.llm_utils as llm_utils
from constants.delimiters import Delimiter
from src.prompts.prompt_templates import invoke_derived_term_date
from src.prompts.fieldset import FieldSet
from src.prompts.prompt_templates import invoke_derived_term_date
from src.utils import string_utils
# Determine the base directory for the project
BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# Path to the mappings folder
fieldExtraction/src/investment/prompt_calls.py
+36 -11
View File
@@ -1,10 +1,11 @@
import logging
import src.config as config
import src.prompts.prompt_templates as prompt_templates
from src.utils import llm_utils, string_utils
from src.prompts.fieldset import FieldSet, Field
from constants.constants import Constants
from constants.delimiters import Delimiter
import src.config as config
from src.prompts.fieldset import Field, FieldSet
from src.utils import llm_utils, string_utils
def prompt_exhibit_level(
@@ -27,6 +28,7 @@ def prompt_exhibit_level(
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
return llm_answer_final
def prompt_exhibit_lesser(
exhibit_text: str,
exhibit_level_answers: dict,
@@ -40,7 +42,9 @@ def prompt_exhibit_lesser(
)
logging.debug(f"LLM raw output for {filename}: {llm_answer_raw}")
try:
exhibit_lesser_of_answer = string_utils.extract_text_from_delimiters(llm_answer_raw, Delimiter.PIPE)
exhibit_lesser_of_answer = string_utils.extract_text_from_delimiters(
llm_answer_raw, Delimiter.PIPE
)
except:
exhibit_lesser_of_answer = "N/A"
exhibit_level_answers["EXHIBIT_LESSER_OF_STATEMENT"] = exhibit_lesser_of_answer
@@ -65,7 +69,10 @@ def prompt_dynamic_primary(
def prompt_reimbursement_primary(
reimbursement_level_fields: FieldSet, exhibit_text: str, constants: Constants, filename: str
reimbursement_level_fields: FieldSet,
exhibit_text: str,
constants: Constants,
filename: str,
) -> list[dict]:
field_prompts = reimbursement_level_fields.print_prompt_dict(constants)
@@ -93,7 +100,10 @@ def prompt_reimbursement_primary(
def prompt_special_case_primary(
special_case_primary_fields: FieldSet, exhibit_text: str, constants: Constants, filename: str
special_case_primary_fields: FieldSet,
exhibit_text: str,
constants: Constants,
filename: str,
):
field_prompts = special_case_primary_fields.print_prompt_dict(constants)
prompt = prompt_templates.SPECIAL_CASE_PRIMARY(exhibit_text, field_prompts)
@@ -229,7 +239,9 @@ def prompt_special_case_breakout(breakout_template, term_answer, filename):
return llm_answer_final
def prompt_lob_relationship(answer_dict: dict, field: str, exhibit_text: str, filename: str):
def prompt_lob_relationship(
answer_dict: dict, field: str, exhibit_text: str, filename: str
):
"""
Helper function to prompt LLM for LOB relationship based on the field and exhibit text.
"""
@@ -249,7 +261,11 @@ def prompt_lob_relationship(answer_dict: dict, field: str, exhibit_text: str, fi
def prompt_special_case_assignment(
exhibit_text: str, reimbursement_dict: dict, special_case_dicts: list, special_case_term: str, filename: str
exhibit_text: str,
reimbursement_dict: dict,
special_case_dicts: list,
special_case_term: str,
filename: str,
):
# If all non-term values are identical across dictionaries, return first dict
@@ -284,13 +300,21 @@ def prompt_special_case_assignment(
return special_case_dicts[0]
def prompt_full_context(contract_text: str, full_context_fields: FieldSet, constants: Constants, filename: str):
def prompt_full_context(
contract_text: str,
full_context_fields: FieldSet,
constants: Constants,
filename: str,
):
prompt_questions = full_context_fields.print_prompt_dict(constants)
full_context_prompt = prompt_templates.ONE_TO_ONE_MULTI_FIELD_TEMPLATE(
context=contract_text[
0 : min(config.MAX_CONTEXT_LENGTH - len(prompt_questions), len(contract_text) - 1)
0 : min(
config.MAX_CONTEXT_LENGTH - len(prompt_questions),
len(contract_text) - 1,
)
],
questions=prompt_questions,
)
@@ -435,6 +459,7 @@ def split_compound_reimbursement_llm(answer_dict: dict, filename: str) -> list[d
)
return [answer_dict] # Return original if error occurs
def prompt_dynamic(text, field_prompts, filename):
"""
Prompts an LLM to process dynamic fields in a given text and extracts structured responses.
@@ -454,4 +479,4 @@ def prompt_dynamic(text, field_prompts, filename):
) # 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)
return llm_answer_final
return llm_answer_final
fieldExtraction/src/investment/row_funcs.py
+2 -3
View File
@@ -1,11 +1,10 @@
import logging
import pandas as pd
import src.investment.prompt_calls as prompt_calls
from constants.constants import Constants
from src.investment import one_to_one_funcs
from src.utils import string_utils
from constants.constants import Constants
import src.investment.prompt_calls as prompt_calls
def combine_one_to_n(
fieldExtraction/src/investment/smart_chunking_funcs.py
-1
View File
@@ -35,7 +35,6 @@ from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
from nltk.tokenize import word_tokenize
from src import config
from src.prompts.fieldset import FieldSet
fieldExtraction/src/investment/table_funcs.py
-1
View File
@@ -4,7 +4,6 @@ import re
from typing import Any, Dict, List, Pattern, Union
import pandas as pd
import src.utils.string_utils as string_utils
from constants.constants import Constants
from src import config
fieldExtraction/src/investment/tin_npi_funcs.py
+15 -5
View File
@@ -234,15 +234,21 @@ def merge_provider_info(one_to_one_results: dict, provider_info: list[dict]) ->
# Deduplicate while preserving order and remove any "UNKNOWN" entries and overlaps between group and other
group_tins = list(dict.fromkeys("|".join(group_tins).split("|")))
other_tins = list(dict.fromkeys("|".join(other_tins).split("|")))
other_tins = [tin for tin in other_tins if tin not in group_tins and tin != "UNKNOWN"] # remove any group TINs from other TINs
other_tins = [
tin for tin in other_tins if tin not in group_tins and tin != "UNKNOWN"
] # remove any group TINs from other TINs
group_npis = list(dict.fromkeys("|".join(group_npis).split("|")))
other_npis = list(dict.fromkeys("|".join(other_npis).split("|")))
other_npis = [npi for npi in other_npis if npi not in group_npis and npi != "UNKNOWN"] # remove any group NPIs from other NPIs
other_npis = [
npi for npi in other_npis if npi not in group_npis and npi != "UNKNOWN"
] # remove any group NPIs from other NPIs
group_names = list(dict.fromkeys("|".join(group_names).split("|")))
other_names = list(dict.fromkeys("|".join(other_names).split("|")))
other_names = [name for name in other_names if name not in group_names and name != "UNKNOWN"] # remove any group Names from other Names
other_names = [
name for name in other_names if name not in group_names and name != "UNKNOWN"
] # remove any group Names from other Names
# Merge group provider information into one_to_one_results
one_to_one_results["PROV_GROUP_TIN"] = "|".join(group_tins) if group_tins else ""
@@ -675,8 +681,12 @@ def run_provider_info_fields(
relevant_pages = [
page_num
for page_num, page_text in text_dict.items()
if any(tin.replace("-", "") in page_text.replace("-", "") for tin in all_tins)
or any(npi.replace("-", "") in page_text.replace("-", "") for npi in all_npis)
if any(
tin.replace("-", "") in page_text.replace("-", "") for tin in all_tins
)
or any(
npi.replace("-", "") in page_text.replace("-", "") for npi in all_npis
)
]
# Stopgap for when the TIN is written in the header or footer of every page (or nearly every page)
fieldExtraction/src/investment/vision_funcs.py
+2 -3
View File
@@ -1,13 +1,12 @@
import base64
from io import BytesIO
import logging
from io import BytesIO
from typing import List
import fitz # PyMuPDF
from PIL import Image # Pillow library
import src.prompts.prompt_templates as prompts
import src.utils.llm_utils as llm_utils
from PIL import Image # Pillow library
from src.config import DOCZY_PDF_FILES_BUCKET_S3_URL
from src.utils import io_utils
fieldExtraction/src/prompts/fieldset.py
+10 -3
View File
@@ -70,7 +70,11 @@ class Field:
self.retrieved_text = None
self.format = field_dict["format"] if "format" in field_dict else None
self.allow_na = field_dict["allow_na"] if "allow_na" in field_dict else True
self.breakout_template = field_dict["breakout_template"] if "breakout_template" in field_dict else None
self.breakout_template = (
field_dict["breakout_template"]
if "breakout_template" in field_dict
else None
)
def __repr__(self):
"""Returns a string representation of the Field object."""
@@ -184,13 +188,16 @@ class Field:
def update_valid_values(self, new_valid):
self.valid_values = str(new_valid)
def get_breakout_template(self):
import src.prompts.prompt_templates as prompt_templates
if hasattr(prompt_templates, self.breakout_template):
return getattr(prompt_templates, self.breakout_template)
else:
raise AttributeError(f"Breakout template '{self.breakout_template}' not found in prompt_templates module")
raise AttributeError(
f"Breakout template '{self.breakout_template}' not found in prompt_templates module"
)
class FieldSet:
fieldExtraction/src/prompts/prompt_templates.py
+78 -18
View File
@@ -2,11 +2,11 @@ import json
from functools import cache
import pandas as pd
import src.config as config
import src.utils.llm_utils as llm_utils
from constants.delimiters import Delimiter
from src.utils.string_utils import extract_text_from_delimiters
from src.prompts.fieldset import FieldSet
import src.config as config
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|."
@@ -15,6 +15,7 @@ PIPE_FORMAT_INSTRUCTIONS = "Briefly explain your answer, then enclose your final
################################## PRIMARY PROMPTS ##################################
#####################################################################################
def EXHIBIT_LEVEL(context, fields):
return f"""Extract attributes from a section of a contract.
@@ -36,6 +37,7 @@ Here is the text to analyze:
Briefly explain your answer, then put the final answer in the properly-formatted JSON dictionary.
"""
def EXHIBIT_LEVEL_LESSER_OF(context):
return f"""Analyze a section of a contract.
@@ -75,6 +77,7 @@ Here is the Exhibit to analyze:
Briefly explain your answer, then put your final answer in |pipes|
"""
def DYNAMIC_PRIMARY_HEADER(context, field_name, field_prompt):
return f"""Extract attributes from the provided text, paying close attention to detail.
@@ -94,6 +97,7 @@ Here is the text to analyze:
Briefly explain your answer before putting the final answer in |pipes|.
"""
def DYNAMIC_PRIMARY_TEXT(context, field_name, field_prompt):
return f"""Extract attributes from the provided text, paying close attention to detail.
@@ -114,6 +118,7 @@ Here is the text to analyze:
Briefly explain your answer before putting the final answer in |pipes|.
"""
def REIMBURSEMENT_PRIMARY(context, fields):
return f"""[OBJECTIVE]
Extract reimbursement attributes from this payer-provider contract section.
@@ -264,6 +269,7 @@ Note: Both services receive the same global cap treatment regardless of table po
Return properly formatted JSON array with all extracted reimbursement items.
"""
def SPECIAL_CASE_PRIMARY(context, fields):
return f"""
[OBJECTIVE]
@@ -308,6 +314,7 @@ Return a properly formatted JSON dictionary containing lists of all extracted re
################################# BREAKOUT PROMPTS ##################################
#####################################################################################
def METHODOLOGY_BREAKOUT(term, questions):
"""
This prompt is used for the fields with field_type="methodology_breakout" + any other fields we need to run based on the result of special case.
@@ -355,6 +362,7 @@ Here is the text to analyze and respond to:
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.
"""
def FEE_SCHEDULE_BREAKOUT(methodology, fee_schedule, questions):
return f"""[OBJECTIVE]
Analyze a given reimbursement methodology from a Payer-Provider contract:
@@ -372,6 +380,7 @@ Methodology: {methodology.replace('"', "'")}
Write your JSON dictionary below:
"""
def GROUPER_BREAKOUT(service, term, questions):
return f"""[OBJECTIVE]
Analyze the given service and reimbursement term to identify grouper-based reimbursement information, and answer the following question(s):
@@ -395,6 +404,7 @@ REIMBURSEMENT TERM: {term}
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 SPECIAL_CASE_BREAKOUT(term, questions):
"""
Generic prompt for special case breakout fields when no additional descriptions are needed.
@@ -426,15 +436,23 @@ def RATE_ESCALATOR_BREAKOUT(term):
Returns:
Formatted prompt string for rate escalator field extraction
"""
questions = FieldSet(config.FIELD_JSON_PATH, field_type="rate_escalator_breakout").print_prompt_dict()
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="rate_escalator_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
def TRIGGER_CAP_BREAKOUT(term):
questions = FieldSet(config.FIELD_JSON_PATH, field_type="trigger_cap_breakout").print_prompt_dict()
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="trigger_cap_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
def OUTLIER_BREAKOUT(term):
questions = FieldSet(config.FIELD_JSON_PATH, field_type="outlier_breakout").print_prompt_dict()
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="outlier_breakout"
).print_prompt_dict()
return 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.
@@ -471,35 +489,53 @@ Briefly explain your answer, then provide the extracted outlier payment informat
"""
def FACILITY_ADJUSTMENT_BREAKOUT(term):
questions = FieldSet(config.FIELD_JSON_PATH, field_type="facility_adjustment_breakout").print_prompt_dict()
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="facility_adjustment_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
def SEQUESTRATION_BREAKOUT(term):
questions = FieldSet(config.FIELD_JSON_PATH, field_type="sequestration_breakout").print_prompt_dict()
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="sequestration_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
def DISCOUNT_BREAKOUT(term):
questions = FieldSet(config.FIELD_JSON_PATH, field_type="discount_breakout").print_prompt_dict()
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="discount_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
def PREMIUM_BREAKOUT(term):
questions = FieldSet(config.FIELD_JSON_PATH, field_type="premium_breakout").print_prompt_dict()
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="premium_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
def ADDITION_BREAKOUT(term):
questions = FieldSet(config.FIELD_JSON_PATH, field_type="addition_breakout").print_prompt_dict()
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="addition_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
def STOP_LOSS_BREAKOUT(term):
questions = FieldSet(config.FIELD_JSON_PATH, field_type="stop_loss_breakout").print_prompt_dict()
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="stop_loss_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
#####################################################################################
################################### CODE PROMPTS ####################################
#####################################################################################
def CODE_EXPLICIT(service, questions):
return f"""Analyze a given medical service:
@@ -520,6 +556,7 @@ Service: {service.replace('"', "'")}
Briefly explain your answer before putting the final answer in JSON dictionary format.
"""
def CODE_CATEGORY(service, choices):
return f"""Analyze a given medical Service Term.
@@ -544,6 +581,7 @@ Here is the Service Term to analyze:
Explain your answer in 1-2 sentences. Write your final answer in JSON list format.
"""
def CODE_IMPLICIT_SPECIAL(service):
return f"""Analyze a given medical Service Term.
@@ -561,6 +599,7 @@ Here is the Service to analyze:
Briefly explain your answer, but put your final answer in |pipes|
"""
def CODE_IMPLICIT(service, choices):
return f"""Analyze a given medical Service Term.
@@ -593,6 +632,7 @@ Here is the Service Term to analyze:
Explain your answer, ensuring each point in the instructions is addressed. Then return your final answer in JSON list format.
"""
def CODE_LAST_CHECK(service):
return f"""Analyze the given healthcare-related term.
@@ -604,6 +644,7 @@ Here is the service to analyze: {service}
Briefly explain your answer, then put your final answer in |pipes|.
"""
def FILL_BILL_TYPE(service, choices):
return f"""Analyze a given medical Service Term.
@@ -633,6 +674,7 @@ Briefly explain your answer, then return your final answer in JSON list format.
################################ ONE-TO-ONE-PROMPTS #################################
#####################################################################################
def ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(context, question):
return f"""The following is a contract (or excerpt thereof). Answer the following question: {question}
@@ -643,6 +685,7 @@ def ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(context, question):
{PIPE_FORMAT_INSTRUCTIONS}
"""
def TIN_NPI_TEMPLATE(context, questions):
return 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.
@@ -695,6 +738,7 @@ Contract text:
{context.replace('"', "'")}
"""
def GROUP_TIN_NPI_TEMPLATE(key_sections: str, provider_list: str) -> str:
return f"""
Identify which provider is the MAIN CONTRACTING GROUP in this contract.
@@ -728,12 +772,14 @@ def GROUP_TIN_NPI_TEMPLATE(key_sections: str, provider_list: str) -> str:
Feel free to justify your answer, but enclose your final answer in |pipes|.
"""
def FULL_CONTEXT_ADDITIONAL_INSTRUCTIONS(allow_na):
if allow_na:
return " Only provide an answer if that answer clearly applies to the ENTIRE contract. Otherwise, answer N/A."
else:
return " Do NOT leave this field N/A. Choose the most appropriate answer based on the context."
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}
@@ -761,10 +807,12 @@ The term length appears in two places but I'm selecting...
}}
"""
#####################################################################################
############################### PREPROCESSING PROMPTS ###############################
#####################################################################################
def EXHIBIT_HEADER(context, EXHIBIT_HEADER_MARKERS):
return f"""Analyze the following contract excerpt and extract the full header found.
@@ -796,6 +844,7 @@ Before returning any output, ensure that all instructions for inclusion were fol
Enclose your final answer in |pipes|, followed by a brief explanation.
"""
def EXHIBIT_LINKAGE(header1, header2):
return f"""
@@ -821,6 +870,7 @@ Briefly explain your answer, then enclose your final answer in |pipes|.
########################## CHECKS, FIXES, AND VALIDATIONS ###########################
#####################################################################################
def VALIDATE_REIMBURSEMENTS_PROMPT(service_term: str, reimb_term: str) -> str:
"""Validates whether a service-reimbursement pair represents a complete reimbursement methodology.
Args:
@@ -869,6 +919,7 @@ Examples:
Briefly explain your reasoning, then return YES or NO in |pipes|."""
def SPLIT_REIMBURSEMENTS_PROMPT(service_term: str, reimb_term: str) -> str:
"""Prompt for splitting a service-reimbursement pair into its component parts.
Args:
@@ -940,6 +991,7 @@ Feel free to justify your answer, but ensure the output is a valid JSON array wi
SERVICE: {service_term}
REIMBURSEMENT: {reimb_term}"""
def DATE_FIX_PROMPT(context: str) -> str:
return f"""Given a date string, convert it to YYYY/MM/DD format following these rules:
Input: Text potentially containing a date
@@ -985,6 +1037,7 @@ Please analyze this date:
{PIPE_FORMAT_INSTRUCTIONS}
"""
def CHECK_REDUNDANT_LESSER(reimb_term, exhibit_lesser_of):
return f"""You are analyzing whether an exhibit-level lesser-of constraint is semantically redundant with a reimbursement methodology.
@@ -1012,6 +1065,7 @@ Answer YES only if the exhibit constraint adds NO additional limitation beyond w
Briefly explain your reasoning, then return YES or NO in |pipes|."""
def CARVEOUT_CHECK(service, methodology, VALID_CARVEOUTS):
carveout_definitions = {
carveout: definition
@@ -1032,6 +1086,7 @@ Determine which case description corresponds to the Service and Methodology comb
{PIPE_FORMAT_INSTRUCTIONS}
"""
def DUAL_LOB_CHECK(service_term):
return f"""Examine the given Service and determine if the Service applies to Medicare, Medicaid, or Both.
@@ -1047,6 +1102,7 @@ Here is the Service:
Explain your answer in one or two sentences, but put your final answer in |pipes|.
"""
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.
Rules:
@@ -1057,6 +1113,7 @@ def EFFECTIVE_DATE_FIX_PROMPT() -> str:
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."""
def SPLIT_REIMB_DATES(date_range: str) -> str:
"""
Splits a date range into start and end dates, returning them in YYYY/MM/DD format.
@@ -1102,6 +1159,7 @@ 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 LOB_RELATIONSHIP(exhibit_text, dynamic_primary_values):
return f"""Analyze the exhibit from a Payer-Provider contract.
@@ -1131,6 +1189,7 @@ Here is the Exhibit to be analyzed:
Briefly explain your reasoning, then put your final answer in |pipes|.
"""
def DERIVED_TERM_DATE_PROMPT(
effective_date: str = None, termination_information: str = None
) -> str:
@@ -1146,6 +1205,7 @@ 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}"
@cache
def invoke_derived_term_date(
effective_date: str = None, termination_information: str = None
@@ -1154,7 +1214,10 @@ def invoke_derived_term_date(
response = llm_utils.invoke_claude(prompt, "haiku_latest", "derive_term_date")
return extract_text_from_delimiters(response, Delimiter.PIPE)
def SPECIAL_CASE_ASSIGNMENT(exhibit_text, reimbursement_dict, special_case_dicts, special_case_term):
def SPECIAL_CASE_ASSIGNMENT(
exhibit_text, reimbursement_dict, special_case_dicts, special_case_term
):
return f"""[OBJECTIVE]
Analyze a section of a contract. Determine which of {special_case_term} values are associated with the REIMBURSEMENT TERM shown.
@@ -1173,6 +1236,3 @@ Return the integer of the {special_case_term} that is most associated with the R
{PIPE_FORMAT_INSTRUCTIONS}
"""
fieldExtraction/src/testbed/add_reimb_id_to_testbed.py
+2 -2
View File
@@ -9,9 +9,9 @@ Run it with `poetry run python -m src.testbed.add_reimb_id_to_testbed`
# Imports
import pandas as pd
from constants.investment_columns import COLUMN_ORDER
from src.investment.postprocessing_funcs import generate_reimb_ids, reorder_columns
from src.investment.postprocessing_funcs import (generate_reimb_ids,
reorder_columns)
# Read the testbed file
testbed = pd.read_excel("Doczy-Testbed.xlsx")
fieldExtraction/src/testbed/one_to_one_comparison.py
-1
View File
@@ -1,5 +1,4 @@
import pandas as pd
from src import config
from src.prompts.fieldset import FieldSet
fieldExtraction/src/testbed/test.py
-2
View File
@@ -2,11 +2,9 @@
import warnings
import pandas as pd
import src.config as config
import src.testbed.testbed_utils as testbed_utils
warnings.filterwarnings("ignore")
########################## READ AND PREPROCESS ##########################
fieldExtraction/src/testbed/testbed_postprocess.py
-1
View File
@@ -12,7 +12,6 @@ output_filename = "src/testbed/v6_DoczyAI_all_test_bed_LIVE_5.6.25_POSTPROCESSED
#### IMPORTS ####
import pandas as pd
from src.testbed.testbed_utils import testbed_postprocess
#### EXECUTION ####
fieldExtraction/src/testbed/testbed_utils.py
+1 -3
View File
@@ -3,12 +3,10 @@ import re
from collections import Counter
import pandas as pd
from rapidfuzz import fuzz
import src.config as config
import src.utils.string_utils as string_utils
from constants.investment_columns import COLUMN_ORDER
from rapidfuzz import fuzz
# Imports for the testbed postprocessing functions
from src.investment import postprocessing_funcs
from src.prompts.fieldset import FieldSet
fieldExtraction/src/tracking/batch_tracking.py
-1
View File
@@ -6,7 +6,6 @@ from pathlib import Path
import boto3
import pandas as pd
from src import config
fieldExtraction/src/tracking/tracking_utils.py
-1
View File
@@ -5,7 +5,6 @@ import time
from datetime import datetime
import pandas as pd
from src import config
fieldExtraction/src/utils/embedding_utils.py
-1
View File
@@ -3,7 +3,6 @@ import pickle
import faiss
import numpy as np
import src.config as config
from src.utils.string_utils import datetime_str
fieldExtraction/src/utils/llm_utils.py
+1 -2
View File
@@ -8,9 +8,8 @@ import time
from collections import OrderedDict
import anthropic
from botocore.exceptions import ClientError, ReadTimeoutError
import src.utils.string_utils as string_utils
from botocore.exceptions import ClientError, ReadTimeoutError
from src import config
fieldExtraction/src/utils/qa_qc_utils.py
+2 -3
View File
@@ -3,11 +3,10 @@ import re
from datetime import datetime
import pandas as pd
from openpyxl import Workbook
from openpyxl.utils.dataframe import dataframe_to_rows
import src.utils.io_utils as io_utils
import src.utils.string_utils as string_utils
from openpyxl import Workbook
from openpyxl.utils.dataframe import dataframe_to_rows
from src import config
from src.investment import preprocessing_funcs
fieldExtraction/src/utils/string_utils.py
+6 -9
View File
@@ -4,18 +4,13 @@ import re
import traceback
import warnings
from datetime import datetime
from typing import Literal, overload
import numpy as np
import pandas as pd
from typing import overload, Literal
from constants.delimiters import Delimiter
from constants.regex_patterns import (
BACKTICK_PATTERN,
PIPE_PATTERN,
TRIPLE_BACKTICK_PATTERN,
)
from constants.regex_patterns import (BACKTICK_PATTERN, PIPE_PATTERN,
TRIPLE_BACKTICK_PATTERN)
def extract_text_from_delimiters(
@@ -227,7 +222,9 @@ def universal_json_load(string_dict: str):
try:
# Clean up the string list before parsing
cleaned_str = re.sub(r'(?<!\\)"', '"', matched_str) # Handle escaped quotes
cleaned_str = re.sub(r"'", '"', cleaned_str) # Replace single quotes with double quotes
cleaned_str = re.sub(
r"'", '"', cleaned_str
) # Replace single quotes with double quotes
return json.loads(cleaned_str)
except json.JSONDecodeError as e:
fieldExtraction/tests/test_code_funcs.py
-1
View File
@@ -4,7 +4,6 @@ import unittest
from unittest.mock import MagicMock, patch
import pandas as pd
import src.codes.code_funcs as code_funcs
from constants.constants import Constants
from constants.delimiters import Delimiter
fieldExtraction/tests/test_crosswalk_utils.py
-1
View File
@@ -2,7 +2,6 @@ import json
import pandas as pd
import pytest
from src.crosswalk.crosswalk_builder import CrosswalkBuilder
from src.utils.crosswalk_utils import apply_crosswalk
fieldExtraction/tests/test_dynamic.py
+221 -114
View File
@@ -1,13 +1,12 @@
import unittest
import json
import os
from unittest.mock import patch, MagicMock
import unittest
from unittest.mock import MagicMock, patch
import pandas as pd
from constants.constants import Constants
from src.investment import dynamic_funcs, prompt_calls
from src.prompts.fieldset import Field, FieldSet
from constants.constants import Constants
class TestDynamicFuncsWithRealConstants(unittest.TestCase):
@@ -17,81 +16,107 @@ class TestDynamicFuncsWithRealConstants(unittest.TestCase):
"""Set up test fixtures before each test method."""
# Create real Constants object
self.constants = Constants()
# Create test fields
self.test_field = Field.from_values(
field_name="TEST_FIELD",
relationship="one-to-one",
field_type="exhibit_level",
prompt="What is the test field?",
base_field="TEST_FIELD_BASE"
base_field="TEST_FIELD_BASE",
)
self.test_field2 = Field.from_values(
field_name="TEST_FIELD2",
relationship="one-to-many",
field_type="exhibit_level",
prompt="What is the second test field?",
base_field="TEST_FIELD2_BASE"
base_field="TEST_FIELD2_BASE",
)
self.escalator_field = Field.from_values(
field_name="RATE_ESCALATOR_DESC",
relationship="one-to-many",
field_type="rate_escalator",
prompt="What is the rate escalator?",
base_field="RATE_ESCALATOR_STATEMENT"
base_field="RATE_ESCALATOR_STATEMENT",
)
self.outlier_field = Field.from_values(
field_name="OUTLIER_TERMS",
relationship="one-to-many",
field_type="exhibit_level",
prompt="What are the outlier terms?",
base_field="OUTLIER_TERMS"
base_field="OUTLIER_TERMS",
)
# Create test FieldSets
self.test_fieldset = FieldSet()
self.test_fieldset.add_field(self.test_field)
self.test_fieldset.add_field(self.test_field2)
# Mock filename
self.filename = "test_file.pdf"
# Sample text
self.exhibit_text = "This is a sample exhibit text with test data."
self.exhibit_header = "This is a sample exhibit header with test data."
# Sample answer dictionaries
self.answer_dicts = [
{"TEST_FIELD": "Value 1", "TEST_FIELD2": "", "RATE_ESCALATOR_DESC": "", "AARETE_DERIVED_LOB": "Commercial"},
{"TEST_FIELD": "Value 2", "TEST_FIELD2": "", "RATE_ESCALATOR_DESC": "", "AARETE_DERIVED_LOB": "Medicare"},
{"TEST_FIELD": "Value 3", "TEST_FIELD2": "", "OUTLIER_TERMS": "", "AARETE_DERIVED_LOB": ""}
{
"TEST_FIELD": "Value 1",
"TEST_FIELD2": "",
"RATE_ESCALATOR_DESC": "",
"AARETE_DERIVED_LOB": "Commercial",
},
{
"TEST_FIELD": "Value 2",
"TEST_FIELD2": "",
"RATE_ESCALATOR_DESC": "",
"AARETE_DERIVED_LOB": "Medicare",
},
{
"TEST_FIELD": "Value 3",
"TEST_FIELD2": "",
"OUTLIER_TERMS": "",
"AARETE_DERIVED_LOB": "",
},
]
@patch('src.utils.llm_utils.invoke_claude')
@patch('src.utils.string_utils.universal_json_load')
@patch("src.utils.llm_utils.invoke_claude")
@patch("src.utils.string_utils.universal_json_load")
def test_prompt_dynamic(self, mock_json_load, mock_invoke_claude):
"""Test the prompt_dynamic function."""
# Setup
field_prompts = {
"TEST_FIELD": "What is the test field?",
"TEST_FIELD2": "What is the second test field?"
"TEST_FIELD2": "What is the second test field?",
}
# Configure mocks
mock_invoke_claude.return_value = '{"TEST_FIELD": ["Value 1"], "TEST_FIELD2": ["Value 2"]}'
mock_json_load.return_value = {"TEST_FIELD": ["Value 1"], "TEST_FIELD2": ["Value 2"]}
mock_invoke_claude.return_value = (
'{"TEST_FIELD": ["Value 1"], "TEST_FIELD2": ["Value 2"]}'
)
mock_json_load.return_value = {
"TEST_FIELD": ["Value 1"],
"TEST_FIELD2": ["Value 2"],
}
# Execute
result = prompt_calls.prompt_dynamic(self.exhibit_text, field_prompts, self.filename)
result = prompt_calls.prompt_dynamic(
self.exhibit_text, field_prompts, self.filename
)
# Assert
mock_invoke_claude.assert_called_once()
mock_json_load.assert_called_once_with('{"TEST_FIELD": ["Value 1"], "TEST_FIELD2": ["Value 2"]}')
self.assertEqual(result, {"TEST_FIELD": ["Value 1"], "TEST_FIELD2": ["Value 2"]})
mock_json_load.assert_called_once_with(
'{"TEST_FIELD": ["Value 1"], "TEST_FIELD2": ["Value 2"]}'
)
self.assertEqual(
result, {"TEST_FIELD": ["Value 1"], "TEST_FIELD2": ["Value 2"]}
)
# Verify that the prompt contains our text
prompt_call = mock_invoke_claude.call_args[0][0]
self.assertIn(self.exhibit_text, prompt_call)
@@ -103,10 +128,12 @@ class TestDynamicFuncsWithRealConstants(unittest.TestCase):
# Setup
one_to_one_fields = FieldSet()
field_to_add = self.test_field
# Execute
result = dynamic_funcs.add_full_context_field(one_to_one_fields, field_to_add, self.answer_dicts)
result = dynamic_funcs.add_full_context_field(
one_to_one_fields, field_to_add, self.answer_dicts
)
# Assert
self.assertEqual(len(result.fields), 1)
self.assertEqual(result.fields[0].field_name, "TEST_FIELD")
@@ -122,12 +149,14 @@ class TestDynamicFuncsWithRealConstants(unittest.TestCase):
relationship="one-to-one",
field_type="exhibit_level",
prompt="What is the program?",
base_field="PROGRAM"
base_field="PROGRAM",
)
# Execute - use our test answer_dicts which has multiple non-empty LOBs
result = dynamic_funcs.add_full_context_field(one_to_one_fields, program_field, self.answer_dicts)
result = dynamic_funcs.add_full_context_field(
one_to_one_fields, program_field, self.answer_dicts
)
# Assert
self.assertEqual(len(result.fields), 0) # Field should not be added
@@ -140,76 +169,102 @@ class TestDynamicFuncsWithRealConstants(unittest.TestCase):
relationship="one-to-one",
field_type="exhibit_level",
prompt="What is the program?",
base_field="PROGRAM"
base_field="PROGRAM",
)
# Use modified answer_dicts with only one non-empty LOB
single_lob_answer_dicts = [
{"TEST_FIELD": "Value 1", "TEST_FIELD2": "", "RATE_ESCALATOR_DESC": "", "AARETE_DERIVED_LOB": "Commercial"},
{"TEST_FIELD": "Value 2", "TEST_FIELD2": "", "RATE_ESCALATOR_DESC": "", "AARETE_DERIVED_LOB": ""},
{"TEST_FIELD": "Value 3", "TEST_FIELD2": "", "OUTLIER_TERMS": "", "AARETE_DERIVED_LOB": ""}
{
"TEST_FIELD": "Value 1",
"TEST_FIELD2": "",
"RATE_ESCALATOR_DESC": "",
"AARETE_DERIVED_LOB": "Commercial",
},
{
"TEST_FIELD": "Value 2",
"TEST_FIELD2": "",
"RATE_ESCALATOR_DESC": "",
"AARETE_DERIVED_LOB": "",
},
{
"TEST_FIELD": "Value 3",
"TEST_FIELD2": "",
"OUTLIER_TERMS": "",
"AARETE_DERIVED_LOB": "",
},
]
# Execute
result = dynamic_funcs.add_full_context_field(one_to_one_fields, program_field, single_lob_answer_dicts)
result = dynamic_funcs.add_full_context_field(
one_to_one_fields, program_field, single_lob_answer_dicts
)
# Assert
self.assertEqual(len(result.fields), 1) # Field should be added
self.assertEqual(result.fields[0].field_name, "PROGRAM")
@patch('src.investment.dynamic_funcs.Field.load_from_file')
@patch('src.investment.dynamic_funcs.FieldSet')
def test_get_dynamic_one_to_one_fields_any_empty(self, mock_fieldset_class, mock_load_field):
@patch("src.investment.dynamic_funcs.Field.load_from_file")
@patch("src.investment.dynamic_funcs.FieldSet")
def test_get_dynamic_one_to_one_fields_any_empty(
self, mock_fieldset_class, mock_load_field
):
"""Test getting dynamic one-to-one fields with any empty criterion."""
# Setup
# Configure mock for fieldset class
mock_all_empty_fields = MagicMock()
mock_all_empty_fields.fields = []
mock_any_empty_fields = MagicMock()
mock_field1 = MagicMock()
mock_field1.field_name = "RATE_ESCALATOR_DESC"
mock_field1.base_field = "RATE_ESCALATOR_STATEMENT"
mock_any_empty_fields.fields = [mock_field1]
mock_one_to_one_fields = MagicMock()
mock_one_to_one_fields.fields = []
mock_fieldset_class.side_effect = [
mock_all_empty_fields,
mock_any_empty_fields,
mock_one_to_one_fields
mock_one_to_one_fields,
]
# Configure mock for loading fields
escalator_statement_field = Field.from_values(
field_name="RATE_ESCALATOR_STATEMENT",
relationship="one-to-one",
field_type="full_context",
prompt="What is the rate escalator statement?"
prompt="What is the rate escalator statement?",
)
mock_load_field.return_value = escalator_statement_field
# Execute
result = dynamic_funcs.get_dynamic_one_to_one_fields(self.answer_dicts)
@patch('src.investment.dynamic_funcs.prompt_calls.prompt_dynamic_primary')
@patch("src.investment.dynamic_funcs.prompt_calls.prompt_dynamic_primary")
def test_dynamic_primary_with_header_answers(self, mock_prompt_dynamic_primary):
"""Test dynamic_primary function with answers in the header."""
# Setup
dynamic_fields = self.test_fieldset
reimbursement_level_fields = FieldSet()
exhibit_level_answers = {}
# Configure mocks
mock_prompt_dynamic_primary.side_effect = ["Single Answer", "Another Answer"]
# Execute
exhibit_level_answers, updated_reimbursement_fields = dynamic_funcs.dynamic_primary(
self.exhibit_text, self.exhibit_header, exhibit_level_answers,
self.constants, self.filename, dynamic_fields, reimbursement_level_fields
exhibit_level_answers, updated_reimbursement_fields = (
dynamic_funcs.dynamic_primary(
self.exhibit_text,
self.exhibit_header,
exhibit_level_answers,
self.constants,
self.filename,
dynamic_fields,
reimbursement_level_fields,
)
)
# Assert
self.assertEqual(len(exhibit_level_answers), 2) # Both fields should be added
self.assertEqual(exhibit_level_answers["TEST_FIELD"], "Single Answer")
@@ -217,110 +272,162 @@ class TestDynamicFuncsWithRealConstants(unittest.TestCase):
mock_prompt_dynamic_primary.assert_called()
self.assertEqual(mock_prompt_dynamic_primary.call_count, 2)
@patch('src.investment.dynamic_funcs.prompt_calls.prompt_dynamic_primary')
def test_dynamic_primary_with_multiple_header_answers(self, mock_prompt_dynamic_primary):
@patch("src.investment.dynamic_funcs.prompt_calls.prompt_dynamic_primary")
def test_dynamic_primary_with_multiple_header_answers(
self, mock_prompt_dynamic_primary
):
"""Test dynamic_primary function with multiple answers in the header."""
# Setup
dynamic_fields = self.test_fieldset
reimbursement_level_fields = FieldSet()
exhibit_level_answers = {}
# Configure mocks to return multiple values for first field
mock_prompt_dynamic_primary.side_effect = ["Value 1|Value 2|Value 3", "Another Answer"]
# Execute
exhibit_level_answers, updated_reimbursement_fields = dynamic_funcs.dynamic_primary(
self.exhibit_text, self.exhibit_header, exhibit_level_answers,
self.constants, self.filename, dynamic_fields, reimbursement_level_fields
)
# Assert
self.assertEqual(len(exhibit_level_answers), 1) # One field should be added to exhibit level
self.assertEqual(exhibit_level_answers["TEST_FIELD2"], "Another Answer")
self.assertEqual(len(updated_reimbursement_fields.fields), 1) # One field should be added to reimbursement level
self.assertEqual(updated_reimbursement_fields.fields[0].field_name, "TEST_FIELD")
self.assertEqual(updated_reimbursement_fields.fields[0].valid_values, "Value 1|Value 2|Value 3")
self.assertTrue("Ensure ALL values that apply" in updated_reimbursement_fields.fields[0].prompt)
@patch('src.investment.prompt_calls.prompt_dynamic')
# Configure mocks to return multiple values for first field
mock_prompt_dynamic_primary.side_effect = [
"Value 1|Value 2|Value 3",
"Another Answer",
]
# Execute
exhibit_level_answers, updated_reimbursement_fields = (
dynamic_funcs.dynamic_primary(
self.exhibit_text,
self.exhibit_header,
exhibit_level_answers,
self.constants,
self.filename,
dynamic_fields,
reimbursement_level_fields,
)
)
# Assert
self.assertEqual(
len(exhibit_level_answers), 1
) # One field should be added to exhibit level
self.assertEqual(exhibit_level_answers["TEST_FIELD2"], "Another Answer")
self.assertEqual(
len(updated_reimbursement_fields.fields), 1
) # One field should be added to reimbursement level
self.assertEqual(
updated_reimbursement_fields.fields[0].field_name, "TEST_FIELD"
)
self.assertEqual(
updated_reimbursement_fields.fields[0].valid_values,
"Value 1|Value 2|Value 3",
)
self.assertTrue(
"Ensure ALL values that apply"
in updated_reimbursement_fields.fields[0].prompt
)
@patch("src.investment.prompt_calls.prompt_dynamic")
def test_dynamic_with_real_constants(self, mock_prompt_dynamic):
"""Test dynamic function with real Constants object."""
# Setup
dynamic_fields = self.test_fieldset
reimbursement_level_fields = FieldSet()
exhibit_level_answers = {}
# Configure mocks
mock_prompt_dynamic.side_effect = [
{"TEST_FIELD": "Header Answer", "TEST_FIELD2": ""}, # Header results
{"TEST_FIELD2": "Text Answer"} # Text results (only for remaining field)
{"TEST_FIELD2": "Text Answer"}, # Text results (only for remaining field)
]
# Execute
exhibit_level_answers, updated_reimbursement_fields = dynamic_funcs.dynamic(
self.exhibit_text, self.exhibit_header, exhibit_level_answers,
self.constants, self.filename, dynamic_fields, reimbursement_level_fields
self.exhibit_text,
self.exhibit_header,
exhibit_level_answers,
self.constants,
self.filename,
dynamic_fields,
reimbursement_level_fields,
)
# Assert
self.assertEqual(len(updated_reimbursement_fields.fields), 2) # Both fields should be added to reimbursement level
self.assertEqual(
len(updated_reimbursement_fields.fields), 2
) # Both fields should be added to reimbursement level
# Check that the first field was processed from header results
self.assertEqual(updated_reimbursement_fields.fields[0].field_name, "TEST_FIELD")
self.assertEqual(updated_reimbursement_fields.fields[0].valid_values, "Header Answer")
self.assertEqual(
updated_reimbursement_fields.fields[0].field_name, "TEST_FIELD"
)
self.assertEqual(
updated_reimbursement_fields.fields[0].valid_values, "Header Answer"
)
# Check that the second field was processed from text results
self.assertEqual(updated_reimbursement_fields.fields[1].field_name, "TEST_FIELD2")
self.assertEqual(updated_reimbursement_fields.fields[1].valid_values, "Text Answer")
self.assertEqual(
updated_reimbursement_fields.fields[1].field_name, "TEST_FIELD2"
)
self.assertEqual(
updated_reimbursement_fields.fields[1].valid_values, "Text Answer"
)
# Verify that prompt_dynamic was called twice
self.assertEqual(mock_prompt_dynamic.call_count, 2)
@patch('src.investment.prompt_calls.prompt_dynamic')
@patch("src.investment.prompt_calls.prompt_dynamic")
def test_dynamic_with_no_fields(self, mock_prompt_dynamic):
"""Test dynamic function with an empty fieldset."""
# Setup
empty_fieldset = FieldSet()
reimbursement_level_fields = FieldSet()
exhibit_level_answers = {}
# Execute
exhibit_level_answers, updated_reimbursement_fields = dynamic_funcs.dynamic(
self.exhibit_text, self.exhibit_header, exhibit_level_answers,
self.constants, self.filename, empty_fieldset, reimbursement_level_fields
self.exhibit_text,
self.exhibit_header,
exhibit_level_answers,
self.constants,
self.filename,
empty_fieldset,
reimbursement_level_fields,
)
# Assert
self.assertEqual(len(exhibit_level_answers), 0)
self.assertEqual(len(updated_reimbursement_fields.fields), 0)
mock_prompt_dynamic.assert_not_called()
@patch('src.investment.prompt_calls.prompt_dynamic')
@patch("src.investment.prompt_calls.prompt_dynamic")
def test_dynamic_with_no_answers(self, mock_prompt_dynamic):
"""Test dynamic function when neither header nor text contains answers."""
# Setup
dynamic_fields = self.test_fieldset
reimbursement_level_fields = FieldSet()
exhibit_level_answers = {}
# Configure mocks - no answers anywhere
mock_prompt_dynamic.side_effect = [
{"TEST_FIELD": "", "TEST_FIELD2": ""}, # Header results
{"TEST_FIELD": "", "TEST_FIELD2": ""} # Text results
{"TEST_FIELD": "", "TEST_FIELD2": ""}, # Text results
]
# Execute
exhibit_level_answers, updated_reimbursement_fields = dynamic_funcs.dynamic(
self.exhibit_text, self.exhibit_header, exhibit_level_answers,
self.constants, self.filename, dynamic_fields, reimbursement_level_fields
self.exhibit_text,
self.exhibit_header,
exhibit_level_answers,
self.constants,
self.filename,
dynamic_fields,
reimbursement_level_fields,
)
# Assert
self.assertEqual(len(updated_reimbursement_fields.fields), 0) # No fields should be added to reimbursement level
self.assertEqual(exhibit_level_answers, {"TEST_FIELD": "", "TEST_FIELD2": ""}) # Both fields should be added to exhibit level
self.assertEqual(
len(updated_reimbursement_fields.fields), 0
) # No fields should be added to reimbursement level
self.assertEqual(
exhibit_level_answers, {"TEST_FIELD": "", "TEST_FIELD2": ""}
) # Both fields should be added to exhibit level
self.assertEqual(mock_prompt_dynamic.call_count, 2)
if __name__ == '__main__':
unittest.main()
if __name__ == "__main__":
unittest.main()
fieldExtraction/tests/test_fieldset.py
+3 -2
View File
@@ -1,10 +1,11 @@
import json
import os
import unittest
from unittest.mock import patch, mock_open
from unittest.mock import mock_open, patch
from src.prompts.fieldset import Field, FieldSet, _field_data_cache, _cache_lock
from constants.constants import Constants
from src.prompts.fieldset import (Field, FieldSet, _cache_lock,
_field_data_cache)
class TestFieldWithRealConstants(unittest.TestCase):
fieldExtraction/tests/test_investment_postprocess.py
+1 -4
View File
@@ -3,7 +3,6 @@ from unittest.mock import MagicMock, patch
import pandas as pd
import pytest
from src.investment.postprocessing_funcs import (
date_postprocess, deduplicate_provider_columns,
flatten_singleton_string_list, format_rate_fields_with_commas,
@@ -95,9 +94,7 @@ class TestPostprocessFunctions(unittest.TestCase):
@patch("src.utils.llm_utils.invoke_claude")
@patch("src.investment.postprocessing_funcs.invoke_derived_term_date")
@patch(
"src.prompts.fieldset.FieldSet.load_from_file"
) # Patch this specific method
@patch("src.prompts.fieldset.FieldSet.load_from_file") # Patch this specific method
def test_date_postprocess(
self, mock_load_from_file, mock_derived_term_date, mock_invoke_claude
):
fieldExtraction/tests/test_io_utils.py
+4 -11
View File
@@ -2,17 +2,10 @@ from io import StringIO
import pandas as pd
import pytest
from src.utils.io_utils import (
filter_already_processed,
list_s3_files,
read_input,
read_input_csv,
read_local,
read_s3,
read_s3_csv,
read_xlsb,
remove_txt_extension,
)
from src.utils.io_utils import (filter_already_processed, list_s3_files,
read_input, read_input_csv, read_local,
read_s3, read_s3_csv, read_xlsb,
remove_txt_extension)
class TestIOUtils:
fieldExtraction/tests/test_one_to_n.py
+356 -233
View File
File diff suppressed because it is too large Load Diff
fieldExtraction/tests/test_preprocessing_funcs.py
-1
View File
@@ -1,5 +1,4 @@
import pytest
from src.investment.preprocessing_funcs import (clean_law_symbols,
filter_quick_review,
remove_page_indicators)
fieldExtraction/tests/test_preprocessing_integration.py
-1
View File
@@ -2,7 +2,6 @@ import unittest
from unittest.mock import patch
import pytest
from constants.constants import Constants
from src.investment.preprocess import clean_tables
fieldExtraction/tests/test_smart_chunking_funcs.py
-1
View File
@@ -1,5 +1,4 @@
import pytest
from src.investment.smart_chunking_funcs import parse_chunk, stitch_chunks
fieldExtraction/tests/test_string_utils.py
+5 -10
View File
@@ -1,19 +1,14 @@
import numpy as np
import pandas as pd
import pytest
import src.utils as utils
import src.utils.llm_utils as llm_utils
from constants.delimiters import Delimiter
from src.utils.string_utils import (
contains_reimbursement,
count_reimbursements_in_exhibit,
extract_signature_page,
extract_text_from_delimiters,
is_empty,
json_parsing_search,
page_key_sort,
)
from src.utils.string_utils import (contains_reimbursement,
count_reimbursements_in_exhibit,
extract_signature_page,
extract_text_from_delimiters, is_empty,
json_parsing_search, page_key_sort)
class TestStringUtils:
fieldExtraction/tests/test_table_funcs.py
-1
View File
@@ -3,7 +3,6 @@ import unittest
from unittest.mock import patch
import pandas as pd
from constants.constants import Constants
from src.investment.table_funcs import (END_MARKER, START_MARKER, Table,
combine_continuous_tables,
fieldExtraction/tests/test_tin_npi_funcs.py
+1 -2
View File
@@ -2,9 +2,8 @@ import json
import re
from unittest.mock import MagicMock, patch
import pytest
import constants.regex_patterns as regex_patterns
import pytest
import src.investment.tin_npi_funcs as tin_npi_funcs
from src.prompts.fieldset import FieldSet