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

Merged in feature/one-to-one-funcs-docs-and-cleanup (pull request #720)

Browse Source 
Feature/one to one funcs docs and cleanup

* Refactor imports and remove unused filter_payer_from_providers function

* Update docstring for date_postprocessing function to clarify processing steps

* Augment docstrings for check_and_update_effective_date and run_full_context_fields

* Enhance docstring for one_to_one_breakout function to clarify processing steps and side effects

* remove unused import and augment docstrings


Approved-by: Katon Minhas
This commit is contained in:
Alex Galarce
2025-10-02 17:02:17 +00:00
committed by Katon Minhas
parent b6f1b8d523
commit 9cadef9795
2 changed files with 264 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fieldExtraction/src/investment/one_to_one_funcs.py
+263 -79
View File
@@ -9,11 +9,13 @@ 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
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.fieldset import Field, FieldSet
from src.prompts.prompt_templates import METHODOLOGY_BREAKOUT
@@ -56,46 +58,6 @@ def extract_global_lesser_of(contract_text: str, filename: str) -> dict:
}
def filter_payer_from_providers(
providers: list[dict], payer_name: str, threshold: int = 80
) -> list[dict]:
"""Filters out providers whose names are too similar to the payer name. Sometimes the LLM pulls the payer as one of the providers, and we want to drop those.
Args:
providers (list[dict]): A list of provider dictionaries to filter.
payer_name (str): The name of the payer to compare against.
threshold (int, optional): The similarity threshold (0-100) for filtering. Defaults to 80.
Returns:
list[dict]: Filtered list of providers
"""
if not payer_name or string_utils.is_empty(payer_name) or payer_name == "UNKNOWN":
return providers # Skip filtering if there's not a valid payer name
filtered_providers = []
for provider in providers:
provider_name = provider.get("NAME", "")
if (
not provider_name
or string_utils.is_empty(provider_name)
or provider_name == "UNKNOWN"
):
filtered_providers.append(provider)
continue
# Calculate similarity score using fuzzy matching
similarity_score = fuzz.ratio(provider_name.lower(), payer_name.lower())
# Keep only providers that are sufficiently different from the payer
if similarity_score < threshold:
filtered_providers.append(provider)
else:
logging.info(
f"Filtered out provider that matched payer: '{provider_name}' similar to '{payer_name}' with score {similarity_score}"
)
return filtered_providers
def run_smart_chunked_fields(
one_to_one_fields: FieldSet,
constants,
@@ -229,17 +191,58 @@ def run_smart_chunked_fields(
def check_and_update_effective_date(answers_dict, text_dict, filename):
"""
Checks and updates the effective date in the answers dictionary. If the effective date is empty,
it attempts to extract the effective date using vision-based LLM or full context processing.
"""Apply two-stage postprocessing to effective date with optional vision-based fallback.
This function performs effective date extraction and validation in three steps:
1. **Text postprocessing**: Normalizes existing text-extracted date using
date_postprocessing() to handle common text variations
(e.g., "January 1, 2024" → "2024-01-01")
2. **Format validation**: Runs validate_and_reformat_date() to ensure ISO (YYYY-MM-DD)
format and valid date values
3. **Vision fallback** (conditional): If config.ENABLE_VISION is True AND steps 1-2
yield an empty result, attempts OCR extraction from the source PDF using Claude
vision API on targeted pages
The vision fallback targets specific pages likely to contain effective date information
(identified via extract_effective_date_pages()) to minimize token usage.
Args:
answers_dict (dict): Dictionary containing extracted answers.
text_dict (dict): Dictionary containing text data, typically organized by pages or sections.
filename (str): The name of the file being processed.
answers_dict (dict): Extraction results containing at least "AARETE_DERIVED_EFFECTIVE_DT"
key. This dictionary is **mutated in place** with the updated effective date.
text_dict (dict[int|str, str]): Page-segmented contract text where keys are page
numbers and values are page content. Used to identify candidate pages for
vision extraction via heuristics (e.g., pages containing "effective" keyword).
filename (str): Source filename (e.g., "contract_001.txt"). Used for:
- Logging context
- Deriving PDF path by replacing .txt extension with .pdf
Returns:
dict: Updated answers dictionary with the effective date.
dict: The mutated answers_dict with "AARETE_DERIVED_EFFECTIVE_DT" updated to:
- A validated ISO date string (e.g., "2024-01-15") if extraction succeeded
- "N/A" if all extraction methods failed
Side Effects:
- Mutates answers_dict["AARETE_DERIVED_EFFECTIVE_DT"] in place
- Makes vision API call to Claude if config.ENABLE_VISION=True and text
extraction fails
- Logs to file-specific log via logging_utils context
Notes:
- Vision extraction requires PDF at:
/root/doczy.ai/fieldExtraction/src/investment/pdf_files/{filename}.pdf
- Vision extraction is functional in local testing but the infrastructure
(namely storing and accessing PDFs via S3) is not yet built out in production
- S3 path support exists but is commented out
- Vision API uses get_image_array_based_answer() which converts PDF pages
to images before LLM processing
Example:
>>> answers = {"AARETE_DERIVED_EFFECTIVE_DT": "Jan 1 2024"}
>>> text = {1: "Agreement effective January 1, 2024", 2: "..."}
>>> result = check_and_update_effective_date(answers, text, "contract.txt")
>>> result["AARETE_DERIVED_EFFECTIVE_DT"]
'2024-01-01'
"""
# Postprocess the effective date
@@ -267,6 +270,7 @@ def check_and_update_effective_date(answers_dict, text_dict, filename):
"/root/doczy.ai/fieldExtraction/src/investment/pdf_files" # local path
)
pdf_path = os.path.join(pdf_base_path, filename.replace("txt", "pdf"))
# TODO: Uncomment if/when S3 infrastructure for PDF storage/retrieval is built out
# pdf_path = os.path.join(DOCZY_PDF_FILES_BUCKET_S3_URL, filename.replace("txt", "pdf")) # s3 path
image_answer_dict = get_image_array_based_answer(
pdf_path,
@@ -287,16 +291,61 @@ def run_full_context_fields(
constants: Constants,
filename: str,
) -> dict[str, str]:
"""
Process fields that require full context and return the extracted answers.
"""Process fields requiring full contract context and return extracted answers with post-processing.
This function orchestrates the extraction and processing of contract-level fields that require
access to the entire contract text (as opposed to smart-chunked or reimbursement-level
fields that use targeted context). The workflow includes:
1. **Field Filtering**: Extracts fields marked as "full_context" or "full_context_separate"
2. **State Optimization**: Skips PAYER_STATE extraction if only one state is configured in the batch
3. **LLM Extraction**: Invokes Claude with full contract context to extract field values
4. **State Normalization**: Converts state names to standardized two-letter abbreviations
(e.g., "California" → "CA", "New York" → "NY")
5. **Special Case Breakout**: Processes any "special_case_primary" fields that require additional
parsing (e.g., complex reimbursement terms)
6. **Separate Field Handling**: Processes "full_context_separate" fields individually
(e.g., GLOBAL_LESSER_OF_STATEMENT extraction)
Args:
one_to_one_fields (FieldSet): FieldSet containing the fields to be processed.
contract_text (str): Full text of contract.
constants (Constants): Constants object
filename (str): Name of the file being processed.
one_to_one_fields (FieldSet): Complete set of one-to-one fields to process. The function
filters this to only "full_context" and "full_context_separate" types.
contract_text (str): Full untruncated text of the contract. This is passed to the LLM
with max context length constraints applied in prompt_calls.prompt_full_context().
constants (Constants): Configuration object containing valid codes and mappings
used during extraction and post-processing.
filename (str): Source filename (e.g., "contract_001.txt").
Returns:
dict[str, str]: Dictionary mapping field names to extracted answers.
dict[str, str]: Dictionary mapping field names to extracted values. Common fields include:
- PAYER_STATE: Two-letter state abbreviation or value from config.STATE[0] if single-state batch
- PROVIDER_STATE: Two-letter state abbreviation
- GLOBAL_LESSER_OF_STATEMENT: Global lesser-of constraint text or "N/A"
- Any special_case_primary fields with their breakout results appended
Side Effects:
- Makes LLM API call via prompt_calls.prompt_full_context()
- May make additional LLM calls for special case breakouts via one_to_one_breakout()
- May make additional LLM call for GLOBAL_LESSER_OF_STATEMENT via extract_global_lesser_of()
- Logs extraction progress and results via logging_utils context
Notes:
- State optimization assumes config.STATE contains the list of states in the current batch
- State normalization uses string_utils.normalize_state_to_abbreviation() which handles
full names, abbreviations, and common misspellings
- Special case breakout is only triggered for fields with field_type="special_case_primary"
- GLOBAL_LESSER_OF_STATEMENT is always processed separately due to its unique extraction logic
- Full context fields are processed together in a single LLM call for efficiency
Example:
>>> fields = FieldSet(relationship="one_to_one", file_path="fields.json")
>>> text = "Agreement between Aetna and Dr. Smith, effective in California..."
>>> constants = Constants(...)
>>> result = run_full_context_fields(fields, text, constants, "contract.txt")
>>> result["PROVIDER_STATE"]
'CA'
>>> result["PAYER_STATE"]
'CA'
"""
full_context_fields = one_to_one_fields.filter(field_type="full_context")
@@ -336,15 +385,76 @@ def run_full_context_fields(
return full_context_answers_dict
def one_to_one_breakout(full_context_answers_dict, filename):
"""
For special_case_primary Fields, run the special case breakout prompt.
def one_to_one_breakout(
full_context_answers_dict: dict[str, str], filename: str
) -> dict[str, str]:
"""Process special case primary fields through LLM-based breakout to extract structured sub-fields.
Special case breakout is a two-stage extraction process used for complex contract-level payment
terms that require additional parsing after initial extraction. For example, a STOP_LOSS_TERM
containing "compensate Provider 120% of Medicaid once claim exceeds $300,000" needs breakout to
extract STOP_LOSS_PCT_RATE_ON_EXCESS_CHARGES=120, STOP_LOSS_FIXED_LOSS_THRESHOLD=$300,000, etc.
This function:
1. Loads all fields marked as field_type="special_case_primary" from the field JSON.
These fields represent already-extracted complex cases that need to be broken out.
2. For each special case field present in the input dictionary:
- Retrieves the field's breakout template (e.g., STOP_LOSS_BREAKOUT, OUTLIER_BREAKOUT)
- Invokes Claude with the breakout template and the field's extracted value
- Parses the LLM response into structured sub-fields
- Updates the input dictionary with the breakout results
3. Returns the mutated dictionary with all breakout results appended
Args:
full_context_answers_dict (dict): Dictionary containing answers from full context processing.
filename (str): The name of the file being processed.
full_context_answers_dict (dict[str, str]): Dictionary containing extracted field values
from full context processing. This dictionary is **mutated in place** with breakout results.
Keys are field names (e.g., "STOP_LOSS_TERM", "OUTLIER_TERM"), values are extracted text.
filename (str): Source filename (e.g., "contract_001.txt"). Used for:
- Logging context
- LLM request attribution
- Error tracking
Returns:
dict: Updated answers dictionary with breakout results.
dict[str, str]: The mutated input dictionary with breakout results added. For example:
Input: {"STOP_LOSS_TERM": "compensate Provider 120% of Medicaid once claim exceeds $300,000"}
Output: {
"STOP_LOSS_TERM": "compensate Provider 120% of Medicaid once claim exceeds $300,000",
"STOP_LOSS_PCT_RATE_ON_EXCESS_CHARGES": "120",
"STOP_LOSS_FIXED_LOSS_THRESHOLD": "$300,000",
"STOP_LOSS_FIRST_DOLLAR_IND": "N"
}
Side Effects:
- Mutates full_context_answers_dict in place
- Makes LLM API call via prompt_calls.prompt_special_case_breakout() for each
special_case_primary field present in the input dictionary
- Logs breakout progress and results via logging_utils context
Notes:
- Only processes fields that are (a) marked as special_case_primary in the field JSON
AND (b) present in the input dictionary with non-empty values
- Common special case fields include: STOP_LOSS_TERM, OUTLIER_TERM, SEQUESTRATION_TERM,
DISCOUNT_TERM, PREMIUM_TERM, RATE_ESCALATOR_TERM, FACILITY_ADJUSTMENT_TERM
- Breakout templates are defined in the Field object's breakout_template attribute
(e.g., "STOP_LOSS_BREAKOUT", "OUTLIER_BREAKOUT")
- If a special case field is missing from the input dict, it is silently skipped
- Breakout results are appended directly to the input dictionary (no namespacing)
- If breakout parsing fails, the error is logged but the function continues processing
remaining fields
Example:
>>> answers = {
... "PAYER_NAME": "Aetna",
... "OUTLIER_TERM": "outlier payment of 80% for charges exceeding $50,000"
... }
>>> result = one_to_one_breakout(answers, "contract.txt")
>>> result.keys()
dict_keys(['PAYER_NAME', 'OUTLIER_TERM', 'OUTLIER_PCT_RATE_ON_EXCESS_CHARGES',
'OUTLIER_FIXED_LOSS_THRESHOLD', 'OUTLIER_FIRST_DOLLAR_IND'])
>>> result["OUTLIER_PCT_RATE_ON_EXCESS_CHARGES"]
'80'
>>> result["OUTLIER_FIXED_LOSS_THRESHOLD"]
'$50,000'
"""
special_case_primary_fields = FieldSet(
config.FIELD_JSON_PATH, field_type="special_case_primary"
@@ -364,19 +474,46 @@ def one_to_one_breakout(full_context_answers_dict, filename):
def create_global_lesser_of_row(
original_row: pd.Series, global_methodology_breakout: dict[str, str]
) -> pd.Series:
"""Create a new row with global lesser of constraint applied to original reimbursement term.
"""Create a duplicate reimbursement row with global lesser-of constraint applied.
Takes an original row that lacks lesser of logic and creates a duplicate with the global lesser of
methodology breakout results applied. Preserves the original REIMB_ID and SERVICE_TERM while updating
reimbursement methodology fields.
This function supports the global lesser-of logic by creating an additional reimbursement
row for services that didn't originally have lesser-of constraints. When a contract has
a global lesser-of statement (e.g., "At no time will reimbursement exceed 100% of billed charges"),
this function creates a new row that applies that constraint to existing reimbursement terms which
lack their own lesser-of logic
The workflow:
1. Copies all fields from the original row (preserving REIMB_ID, SERVICE_TERM, etc.)
2. Overwrites methodology fields with values from global_methodology_breakout
3. Sets LESSER_OF_IND="Y" to indicate lesser-of logic applies
4. Adds GLOBAL_LESSER_OF_APPLIED="Y" flag for tracking
This allows the final output to have two rows for the same service:
- Original row: Standard reimbursement without lesser-of constraint
- New row: Same service with global lesser-of constraint applied
Args:
original_row (pd.Series): Original reimbursement row with LESSER_OF_IND="N"
global_methodology_breakout (dict[str, str]): Results from global lesser of methodology breakout,
containing fields like REIMB_FEE_RATE, REIMB_PCT_RATE, etc.
original_row (pd.Series): Original reimbursement row with LESSER_OF_IND="N".
Contains fields like REIMB_ID, SERVICE_TERM, REIMB_PCT_RATE, etc.
global_methodology_breakout (dict[str, str]): Results from
run_global_lesser_of_breakout() containing methodology fields to apply.
Common fields include: REIMB_PCT_RATE, REIMB_FEE_RATE, UNIT_OF_MEASURE, etc.
Returns:
pd.Series: New reimbursement row with global lesser of constraint applied.
pd.Series: New reimbursement row with:
- All original fields copied (REIMB_ID, SERVICE_TERM, etc. unchanged)
- Methodology fields updated with global_methodology_breakout values
- LESSER_OF_IND="Y"
- GLOBAL_LESSER_OF_APPLIED="Y"
Side Effects:
None - creates a new Series without mutating the input
Notes:
- The function does a shallow copy of the original row
- Only fields present in global_methodology_breakout are overwritten
- This function is typically called from row_funcs.apply_global_lesser_of()
which handles the iteration over all eligible rows
"""
new_row = original_row.copy()
@@ -395,18 +532,65 @@ def create_global_lesser_of_row(
def run_global_lesser_of_breakout(
global_lesser_of_stmt: str, constants: Constants, filename: str
) -> dict[str, str]:
"""Run methodology breakout on global lesser of statement to extract structured reimbursement details.
"""Parse global lesser-of statement into structured methodology fields using LLM breakout.
Uses the same REIMB_TERM_BREAKOUT prompt as the one used for individual service terms
to parse the global lesser of statement and extract fields like REIMB_FEE_RATE, REIMB_PCT_RATE, etc.
This function reuses the same METHODOLOGY_BREAKOUT prompt template used for individual
service-level reimbursement terms to extract structured fields from contract-wide lesser-of
constraints. The extracted fields follow the same structure as methodology_breakout fields
defined in the field JSON.
The function works by:
1. Creating a mock service term "Global Lesser Of (all services)" to fit the breakout template
2. Loading methodology_breakout fields from the field JSON configuration
3. Invoking Claude with the METHODOLOGY_BREAKOUT prompt to parse the statement
4. Adding "[GLOBAL LESSER OF]" prefix to REIMB_TERM for tracking/identification
5. Returning the parsed fields or a minimal dict with just REIMB_TERM if parsing fails
Args:
global_lesser_of_stmt (str): Global lesser of statement extracted from contract (e.g., "lesser of billed or allowable").
filename (str): Name of the file being processed, used for logging and debugging.
global_lesser_of_stmt (str): Global lesser-of statement extracted from contract.
Example: "At no time will reimbursement exceed 100% of billed charges"
constants (Constants): Configuration object containing valid codes and business rules
used during methodology breakout parsing
filename (str): Source filename (e.g., "contract_001.txt"). Used for:
- Logging context
- LLM request attribution
- Error tracking
Returns:
dict[str, str]: Dictionary of methodology breakout fields (REIMB_PCT_RATE,
REIMB_FEE_RATE, etc.) extracted from the global lesser of statement.
Returns an empty dict if parsing fails.
dict[str, str]: Dictionary of methodology breakout fields extracted from the statement.
Always includes REIMB_TERM with "[GLOBAL LESSER OF]" prefix. Other fields depend on
what's defined as field_type="methodology_breakout" in the field JSON (e.g.,
REIMB_PCT_RATE, REIMB_FEE_RATE, UNIT_OF_MEASURE, etc.)
If parsing fails, returns: {"REIMB_TERM": "[GLOBAL LESSER OF] {original_statement}"}
Side Effects:
- Makes LLM API call via llm_utils.invoke_claude()
- Logs debug information about prompt and response (via logging_utils context)
- Logs error if parsing fails but continues execution
Notes:
- Uses the same METHODOLOGY_BREAKOUT template as one-to-n service-level terms
- The mock service term "Global Lesser Of (all services)" allows the global statement
to fit the [service, term] tuple expected by METHODOLOGY_BREAKOUT
- Parsing failures are non-fatal - returns minimal dict to allow processing to continue
- The "[GLOBAL LESSER OF]" prefix in REIMB_TERM distinguishes these from service-level terms
- This function is typically called from row_funcs.apply_global_lesser_of() which uses
the results to create duplicate rows via create_global_lesser_of_row()
- The specific methodology fields returned depend on the field_type="methodology_breakout"
definitions in the field JSON configuration
Example:
>>> stmt = "At no time will reimbursement exceed 100% of billed charges"
>>> constants = Constants(...)
>>> result = run_global_lesser_of_breakout(stmt, constants, "contract.txt")
>>> result["REIMB_TERM"]
'[GLOBAL LESSER OF] At no time will reimbursement exceed 100% of billed charges'
>>> result["AARETE_DERIVED_REIMB_METHOD"]
'billed charges'
>>> result["REIMB_PCT_RATE"]
'100'
>>> # Other fields depend on methodology_breakout field definitions in JSON
"""
# Create a mock service term for the breakout
fieldExtraction/src/investment/postprocessing_funcs.py
+1 -1
View File
@@ -147,7 +147,7 @@ def validate_and_reformat_date(date_str: str) -> str:
@cache # memoize repeated calls to this function
def date_postprocessing(date: str) -> str:
"""Processes dates using `investment_prompts.date_fix_prompt`. Following the call to the LLM, the answer is extracted from the response and returned.
"""Processes dates using `investment_prompts.date_fix_prompt`.
Args:
date (str): Input date string to be processed.