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

Merged in bugfix/parser-downstream-improvements (pull request #875)

Browse Source 
Bugfix/parser downstream improvements

* Refactor: Implement field-aware JSON parsers with centralized normalization

This refactor introduces a robust system for normalizing LLM output based on
field format mappings, ensuring consistent data types throughout the pipeline.

Key Changes:
- Add FIELD_FORMAT_MAPPING constant defining expected formats for all fields
- Create format_normalization.py utility for type-aware normalization
- Update json_utils.py parsers to accept field_names/field_name parameters
- Refactor prompt_templates.py to use parser factories (_create_json_dict_parser,
  _create_json_list_parser) that bind field metadata for automatic normalization
- Update prompt_calls.py to pass field names to parsers, eliminating redundant
  normalization logic
- Remove parse_json_dict_or_list (unused, ambiguous function)
- Simplify METHODOLOGY_BREAKOUT and REIMBURSEMENT_PRIMARY to use helper functions
- Add comprehensive integration tests verifying normalization works end-to-end

Benefits:
- Single source of truth for field formats (FIELD_FORMAT_…
* refactor: normalize helper prompt outputs at prompt_calls level

- Update CARVEOUT_CHECK to use field-aware parser for CARVEOUT_CD normalization
- Update LOB_RELATIONSHIP to normalize to string format in prompt_calls.py
- Update SPLIT_REIMB_DATES to normalize date values in prompt_calls.py
- Remove defensive normalization from one_to_n_funcs.py for LOB relationships
- Remove manual normalization from split_reimb_dates() - values now normalized upstream
- All helper prompts that populate fields now normalize at prompt_calls.py level
- Downstream functions receive correctly formatted values without additional processing

* refactor: remove band-aid normalization functions and migrate HSC to field-aware parsers

- Update ONE_TO_ONE_SINGLE_FIELD_TEMPLATE to use field-aware parser with field_name parameter
- Remove list wrapping logic in hybrid_smart_chunking_funcs (field-aware parser handles normalization)
- Remove normalize_one_to_one_field_value and normalize_one_to_one_answers_dict from string_utils.py
- Remove all debug print statements from HSC processing
- Remove commented-out normalization calls from client-specific files (clover, bcbs_promise)
- All normalization now handled exclusively through FIELD_FORMAT_MAPPING via field-aware parsers

* Ran Black for formatting

* Print Statements removed, more cleaning

* Merge branch 'DEV' into bugfix/parser-downstream-improvements

* refactor: combine FIELD_FORMAT_MAPPING into investment_columns.py

- Merged field_format_mapping.py into investment_columns.py to create single source of truth
- FIELD_FORMAT_MAPPING now ordered by COLUMN_ORDER (161 fields)
- Added 5 missing fields from COLUMN_ORDER with default format types
- Updated all imports across codebase to use investment_columns
- Python dict preserves insertion order (3.7+), maintaining COLUMN_ORDER sequence
- All tests passing (38 field-aware tests verified)

* Deprecate COLUMN_ORDER, rely on Mapping only

* Merged DEV into bugfix/parser-downstream-improvements


Approved-by: Katon Minhas
This commit is contained in:
Praneel Panchigar
2026-02-09 22:06:21 +00:00
committed by Katon Minhas
parent 4587042c43
commit 9b0a344b13
21 changed files with 2291 additions and 755 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/constants/investment_columns.py
+185 -164
View File
@@ -1,164 +1,185 @@
# This is the ONE AND ONLY source of investment column names and order. Do not reference anything else
COLUMN_ORDER = [
"AARETE_DERIVED_SID",
"CONTRACT_CLASS",
"AARETE_DERIVED_CONTRACT_CLASS",
"FILE_NAME",
"CONTRACT_TITLE",
"CONTRACT_AMENDMENT_NUM",
"FILENAME_AMENDMENT_NUM",
"AARETE_DERIVED_AMENDMENT_NUM",
"CLIENT_NAME",
"PAYER_NAME",
"PAYER_STATE",
"PROVIDER_STATE",
"FILENAME_TIN",
"PROV_GROUP_TIN",
"PROV_GROUP_NPI",
"PROV_GROUP_NAME_FULL",
"PROV_OTHER_TIN",
"PROV_OTHER_NPI",
"PROV_OTHER_NAME_FULL",
"PROV_INFO_JSON",
"EFFECTIVE_DT",
"AARETE_DERIVED_EFFECTIVE_DT",
"TERMINATION_DT",
"AARETE_DERIVED_TERMINATION_DT",
"AUTO_RENEWAL_IND",
"AUTO_RENEWAL_TERM",
"NUM_AVAILABLE_SIGNATORY_LINE_COUNT",
"NUM_SIGNED_SIGNATORY_LINE_COUNT",
"AARETE_DERIVED_SIGNATORY_COMPLETE_IND",
"EXHIBIT_TITLE",
"EXHIBIT_PAGE",
"REIMB_PROV_TIN",
"REIMB_PROV_NPI",
"REIMB_PROV_NAME",
"REIMB_EFFECTIVE_DT",
"REIMB_TERMINATION_DT",
"CLAIM_TYPE_CD",
"AARETE_DERIVED_CLAIM_TYPE_CD",
"PRODUCT",
"AARETE_DERIVED_PRODUCT",
"LOB",
"AARETE_DERIVED_LOB",
"PROGRAM",
"AARETE_DERIVED_PROGRAM",
"NETWORK",
"AARETE_DERIVED_NETWORK",
"LOB_PROGRAM_RELATIONSHIP",
"LOB_PRODUCT_RELATIONSHIP",
"SERVICE_AREA",
"AARETE_DERIVED_SERVICE_AREA",
"PROV_TYPE",
"AARETE_DERIVED_PROV_TYPE",
"PROV_TAXONOMY_CD",
"PROV_TAXONOMY_CD_DESC",
"PROV_SPECIALTY_CD",
"PROV_SPECIALTY_CD_DESC",
"PLACE_OF_SERVICE_CD",
"PLACE_OF_SERVICE_CD_DESC",
"BILL_TYPE_CD",
"BILL_TYPE_CD_DESC",
"PATIENT_AGE_MIN",
"PATIENT_AGE_MAX",
"REIMB_TERM",
"REIMB_ID",
"REIMB_LESSER_OF_ID",
"CARVEOUT_IND",
"CARVEOUT_CD",
"LESSER_OF_IND",
"GREATER_OF_IND",
"AARETE_DERIVED_REIMB_METHOD",
"UNIT_OF_MEASURE",
"REIMB_PCT_RATE",
"REIMB_FEE_RATE",
"REIMB_CONVERSION_FACTOR",
"TRIGGER_CAP_THRESHOLD_AMT",
"TRIGGER_BASE_THRESHOLD",
"DEFAULT_IND",
"ADDITION_DESC",
"ADDITION_MAX_PCT_RATE_INC",
"ADDITION_MAX_FEE_RATE_INC",
"ADDITION_RATE_CHANGE_TIMELINE",
"AARETE_DERIVED_ADDITION_RATE_CHANGE_TIMELINE",
"FEE_SCHEDULE",
"AARETE_DERIVED_FEE_SCHEDULE",
"FEE_SCHEDULE_VERSION",
"AARETE_DERIVED_FEE_SCHEDULE_VERSION",
"SERVICE_TERM",
"CPT4_PROC_CD",
"CPT4_PROC_CD_DESC",
"CPT4_PROC_MOD",
"CPT4_PROC_MOD_DESC",
"REVENUE_CD",
"REVENUE_CD_DESC",
"DIAG_CD",
"DIAG_CD_DESC",
"NDC_CD",
"NDC_CD_DESC",
"HIPPS_CD",
"HIPPS_CD_DESC",
"RUG_CD",
"RUG_CD_DESC",
"CLAIM_ADMIT_TYPE_CD",
"AUTH_ADMIT_TYPE_DESC",
"CLAIM_STATUS_CD",
"CLAIM_STATUS_CD_DESC",
"CODE_METHODOLOGY",
"GROUPER_TYPE",
"GROUPER_CD",
"GROUPER_CD_DESC",
"GROUPER_PCT_RATE",
"GROUPER_BASE_RATE",
"GROUPER_VERSION",
"AARETE_DERIVED_GROUPER_VERSION",
"GROUPER_ALTERNATIVE_LEVEL_OF_CARE",
"GROUPER_SEVERITY_IND",
"GROUPER_SEVERITY",
"GROUPER_RISK_OF_MORTALITY_SUBCLASS",
"GROUPER_TRANSFER_IND",
"GROUPER_READMISSIONS_IND",
"GROUPER_HAC_IND",
"OUTLIER_TERM",
"OUTLIER_FIRST_DOLLAR_IND",
"RANGE_NBR_DAYS",
"OUTLIER_FIXED_LOSS_NBR_DAYS_THRESHOLD",
"OUTLIER_FIXED_LOSS_THRESHOLD",
"OUTLIER_MAXIMUM",
"OUTLIER_MAXIMUM_FREQUENCY",
"OUTLIER_PCT_RATE",
"OUTLIER_EXCLUSION_CD",
"OUTLIER_EXCLUSION_CD_DESC",
"FACILITY_ADJUSTMENT_TERM",
"DSH_IND",
"IME_IND",
"NTAP_IND",
"UC_IND",
"GME_IND",
"RATE_ESCALATOR_IND",
"RATE_ESCALATOR_TERM",
"RATE_ESCALATOR_MAX_RATE_INC_PCT",
"RATE_ESCALATOR_RATE_CHANGE_TIMELINE",
"STOP_LOSS_TERM",
"STOP_LOSS_FIRST_DOLLAR_IND",
"STOP_LOSS_RANGE_NBR_DAYS",
"STOP_LOSS_FIXED_LOSS_THRESHOLD",
"STOP_LOSS_MAXIMUM",
"STOP_LOSS_MAXIMUM_FREQUENCY",
"STOP_LOSS_DAILY_MAX_RATE",
"STOP_LOSS_PCT_RATE_ON_EXCESS_CHARGES",
"STOP_LOSS_EXCLUSION_CD",
"STOP_LOSS_EXCLUSION_CD_DESC",
"DISCOUNT_PERCENT_RATE",
"DISCOUNT_RATE_CHANGE_INTERVAL",
"DISCOUNT_START_DT",
"DISCOUNT_END_DT",
"PREMIUM_PERCENT_RATE",
"PREMIUM_RATE_CHANGE_INTERVAL",
"PREMIUM_START_DT",
"PREMIUM_END_DT",
"SEQUESTRATION_TERM",
"SEQUESTRATION_START_DT",
"SEQUESTRATION_END_DT",
]
# This is the ONE AND ONLY source of investment column names, order, and data types.
# Do not reference anything else for column definitions.
"""
Investment column configuration.
This module defines:
- FIELD_FORMAT_MAPPING: Dictionary mapping field names to their expected output format types
Format types:
- 'str': Single string value
- 'list[str]': List of strings
- 'list[dict[str,str]]': List of dictionaries with string keys and values
- 'dict[str,str]': Dictionary with string keys and values
This file should be edited directly. When adding new fields:
1. Add the field name and format type to FIELD_FORMAT_MAPPING
2. Python dicts preserve insertion order (since Python 3.7), so FIELD_FORMAT_MAPPING
will maintain the order defined
"""
# Field format mapping
# Python dicts preserve insertion order (since Python 3.7)
FIELD_FORMAT_MAPPING = {
"AARETE_DERIVED_SID": "str",
"CONTRACT_CLASS": "str",
"AARETE_DERIVED_CONTRACT_CLASS": "str",
"FILE_NAME": "str",
"CONTRACT_TITLE": "str",
"CONTRACT_AMENDMENT_NUM": "str",
"FILENAME_AMENDMENT_NUM": "str", # Default: not in original mapping
"AARETE_DERIVED_AMENDMENT_NUM": "str",
"CLIENT_NAME": "str",
"PAYER_NAME": "str",
"PAYER_STATE": "list[str]",
"PROVIDER_STATE": "list[str]",
"FILENAME_TIN": "list[str]",
"PROV_GROUP_TIN": "list[str]",
"PROV_GROUP_NPI": "list[str]",
"PROV_GROUP_NAME_FULL": "list[str]",
"PROV_OTHER_TIN": "list[str]",
"PROV_OTHER_NPI": "list[str]",
"PROV_OTHER_NAME_FULL": "list[str]",
"PROV_INFO_JSON": "list[dict[str,str]]",
"EFFECTIVE_DT": "str",
"AARETE_DERIVED_EFFECTIVE_DT": "str",
"TERMINATION_DT": "str",
"AARETE_DERIVED_TERMINATION_DT": "str",
"AUTO_RENEWAL_IND": "str",
"AUTO_RENEWAL_TERM": "str",
"NUM_AVAILABLE_SIGNATORY_LINE_COUNT": "str",
"NUM_SIGNED_SIGNATORY_LINE_COUNT": "str",
"AARETE_DERIVED_SIGNATORY_COMPLETE_IND": "str",
"EXHIBIT_TITLE": "str",
"EXHIBIT_PAGE": "str",
"REIMB_PROV_TIN": "list[str]",
"REIMB_PROV_NPI": "list[str]",
"REIMB_PROV_NAME": "list[str]",
"REIMB_EFFECTIVE_DT": "str",
"REIMB_TERMINATION_DT": "str",
"CLAIM_TYPE_CD": "str",
"AARETE_DERIVED_CLAIM_TYPE_CD": "str",
"PRODUCT": "list[str]",
"AARETE_DERIVED_PRODUCT": "list[str]",
"LOB": "list[str]",
"AARETE_DERIVED_LOB": "list[str]",
"PROGRAM": "list[str]",
"AARETE_DERIVED_PROGRAM": "list[str]",
"NETWORK": "list[str]",
"AARETE_DERIVED_NETWORK": "list[str]",
"LOB_PROGRAM_RELATIONSHIP": "str",
"LOB_PRODUCT_RELATIONSHIP": "str",
"SERVICE_AREA": "str", # Default: not in original mapping
"AARETE_DERIVED_SERVICE_AREA": "str", # Default: not in original mapping
"PROV_TYPE": "str", # Default: not in original mapping
"AARETE_DERIVED_PROV_TYPE": "str", # Default: not in original mapping
"PROV_TAXONOMY_CD": "list[str]",
"PROV_TAXONOMY_CD_DESC": "list[str]",
"PROV_SPECIALTY_CD": "list[str]",
"PROV_SPECIALTY_CD_DESC": "list[str]",
"PLACE_OF_SERVICE_CD": "list[str]",
"PLACE_OF_SERVICE_CD_DESC": "list[str]",
"BILL_TYPE_CD": "list[str]",
"BILL_TYPE_CD_DESC": "list[str]",
"PATIENT_AGE_MIN": "str",
"PATIENT_AGE_MAX": "str",
"REIMB_TERM": "str",
"REIMB_ID": "str",
"REIMB_LESSER_OF_ID": "str",
"CARVEOUT_IND": "str",
"CARVEOUT_CD": "str",
"LESSER_OF_IND": "str",
"GREATER_OF_IND": "str",
"AARETE_DERIVED_REIMB_METHOD": "str",
"UNIT_OF_MEASURE": "str",
"REIMB_PCT_RATE": "str",
"REIMB_FEE_RATE": "str",
"REIMB_CONVERSION_FACTOR": "str",
"TRIGGER_CAP_THRESHOLD_AMT": "str",
"TRIGGER_BASE_THRESHOLD": "str",
"DEFAULT_IND": "str",
"ADDITION_DESC": "str",
"ADDITION_MAX_PCT_RATE_INC": "str",
"ADDITION_MAX_FEE_RATE_INC": "str",
"ADDITION_RATE_CHANGE_TIMELINE": "str",
"AARETE_DERIVED_ADDITION_RATE_CHANGE_TIMELINE": "str",
"FEE_SCHEDULE": "str",
"AARETE_DERIVED_FEE_SCHEDULE": "str",
"FEE_SCHEDULE_VERSION": "str",
"AARETE_DERIVED_FEE_SCHEDULE_VERSION": "str",
"SERVICE_TERM": "str",
"CPT4_PROC_CD": "list[str]",
"CPT4_PROC_CD_DESC": "list[str]",
"CPT4_PROC_MOD": "list[str]",
"CPT4_PROC_MOD_DESC": "list[str]",
"REVENUE_CD": "list[str]",
"REVENUE_CD_DESC": "list[str]",
"DIAG_CD": "list[str]",
"DIAG_CD_DESC": "list[str]",
"NDC_CD": "list[str]",
"NDC_CD_DESC": "list[str]",
"HIPPS_CD": "list[str]",
"HIPPS_CD_DESC": "list[str]",
"RUG_CD": "list[str]",
"RUG_CD_DESC": "list[str]",
"CLAIM_ADMIT_TYPE_CD": "list[str]",
"AUTH_ADMIT_TYPE_DESC": "list[str]",
"CLAIM_STATUS_CD": "list[str]",
"CLAIM_STATUS_CD_DESC": "list[str]",
"CODE_METHODOLOGY": "str",
"GROUPER_TYPE": "str",
"GROUPER_CD": "list[str]",
"GROUPER_CD_DESC": "list[str]",
"GROUPER_PCT_RATE": "str",
"GROUPER_BASE_RATE": "str",
"GROUPER_VERSION": "str",
"AARETE_DERIVED_GROUPER_VERSION": "str",
"GROUPER_ALTERNATIVE_LEVEL_OF_CARE": "str",
"GROUPER_SEVERITY_IND": "str",
"GROUPER_SEVERITY": "str",
"GROUPER_RISK_OF_MORTALITY_SUBCLASS": "str",
"GROUPER_TRANSFER_IND": "str",
"GROUPER_READMISSIONS_IND": "str",
"GROUPER_HAC_IND": "str",
"OUTLIER_TERM": "str",
"OUTLIER_FIRST_DOLLAR_IND": "str",
"RANGE_NBR_DAYS": "str",
"OUTLIER_FIXED_LOSS_NBR_DAYS_THRESHOLD": "str",
"OUTLIER_FIXED_LOSS_THRESHOLD": "str",
"OUTLIER_MAXIMUM": "str",
"OUTLIER_MAXIMUM_FREQUENCY": "str",
"OUTLIER_PCT_RATE": "str",
"OUTLIER_EXCLUSION_CD": "list[str]",
"OUTLIER_EXCLUSION_CD_DESC": "list[str]",
"FACILITY_ADJUSTMENT_TERM": "str",
"DSH_IND": "str",
"IME_IND": "str",
"NTAP_IND": "str",
"UC_IND": "str",
"GME_IND": "str",
"RATE_ESCALATOR_IND": "str",
"RATE_ESCALATOR_TERM": "str",
"RATE_ESCALATOR_MAX_RATE_INC_PCT": "str",
"RATE_ESCALATOR_RATE_CHANGE_TIMELINE": "str",
"STOP_LOSS_TERM": "str",
"STOP_LOSS_FIRST_DOLLAR_IND": "str",
"STOP_LOSS_RANGE_NBR_DAYS": "str",
"STOP_LOSS_FIXED_LOSS_THRESHOLD": "str",
"STOP_LOSS_MAXIMUM": "str",
"STOP_LOSS_MAXIMUM_FREQUENCY": "str",
"STOP_LOSS_DAILY_MAX_RATE": "str",
"STOP_LOSS_PCT_RATE_ON_EXCESS_CHARGES": "str",
"STOP_LOSS_EXCLUSION_CD": "list[str]",
"STOP_LOSS_EXCLUSION_CD_DESC": "list[str]",
"DISCOUNT_PERCENT_RATE": "str",
"DISCOUNT_RATE_CHANGE_INTERVAL": "str",
"DISCOUNT_START_DT": "str",
"DISCOUNT_END_DT": "str",
"PREMIUM_PERCENT_RATE": "str",
"PREMIUM_RATE_CHANGE_INTERVAL": "str",
"PREMIUM_START_DT": "str",
"PREMIUM_END_DT": "str",
"SEQUESTRATION_TERM": "str",
"SEQUESTRATION_START_DT": "str",
"SEQUESTRATION_END_DT": "str",
}
src/pipelines/clients/bcbs_promise/file_processing.py
-8
View File
@@ -260,14 +260,6 @@ def run_one_to_one_prompts(
one_to_one_results, contract_text
)
################## Normalize before crosswalk ##################
# Normalize before get_crosswalk_fields (which expects/returns a list)
# get_crosswalk_fields and fill_na_mapping are mechanical steps (no LLM),
# so they won't introduce single-element lists that need normalization
one_to_one_results = string_utils.normalize_one_to_one_answers_dict(
one_to_one_results
)
################## Crosswalk Fields ##################
one_to_one_results = aarete_derived.get_crosswalk_fields(
[one_to_one_results], constants
src/pipelines/clients/bcbs_promise/prompts/prompt_calls.py
-14
View File
@@ -28,15 +28,6 @@ def prompt_exhibit_level(
llm_answer_final = _parser(llm_answer_raw)
# Normalize CLAIM_TYPE_CD if present: convert single-element lists to strings
# This field should be a single value even though it's a 1:N field
if "CLAIM_TYPE_CD" in llm_answer_final:
llm_answer_final["CLAIM_TYPE_CD"] = (
string_utils.normalize_one_to_one_field_value(
"CLAIM_TYPE_CD", llm_answer_final["CLAIM_TYPE_CD"]
)
)
return llm_answer_final
@@ -445,11 +436,6 @@ def prompt_full_context(
instruction=prompt_templates.ONE_TO_ONE_MULTI_FIELD_INSTRUCTION(),
)
full_context_answers_dict = _parser(claude_answer_raw)
# Normalize 1:1 fields: convert single-element lists to strings
# Preserves legitimate multi-value lists (e.g., LOB, PROGRAM when passed to 1:1)
full_context_answers_dict = string_utils.normalize_one_to_one_answers_dict(
full_context_answers_dict
)
except Exception as e:
logging.error(f"Error processing high accuracy fields: {str(e)}")
full_context_answers_dict = {}
src/pipelines/clients/clover/file_processing.py
-8
View File
@@ -262,14 +262,6 @@ def run_one_to_one_prompts(
one_to_one_results, contract_text
)
################## Normalize before crosswalk ##################
# Normalize before get_crosswalk_fields (which expects/returns a list)
# get_crosswalk_fields and fill_na_mapping are mechanical steps (no LLM),
# so they won't introduce single-element lists that need normalization
one_to_one_results = string_utils.normalize_one_to_one_answers_dict(
one_to_one_results
)
################## Crosswalk Fields ##################
one_to_one_results = aarete_derived.get_crosswalk_fields(
[one_to_one_results], constants
src/pipelines/clients/clover/prompts/prompt_calls.py
-15
View File
@@ -34,15 +34,6 @@ def prompt_exhibit_level(
llm_answer_final = _parser(llm_answer_raw)
# Normalize CLAIM_TYPE_CD if present: convert single-element lists to strings
# This field should be a single value even though it's a 1:N field
if "CLAIM_TYPE_CD" in llm_answer_final:
llm_answer_final["CLAIM_TYPE_CD"] = (
string_utils.normalize_one_to_one_field_value(
"CLAIM_TYPE_CD", llm_answer_final["CLAIM_TYPE_CD"]
)
)
return llm_answer_final
@@ -470,12 +461,6 @@ def prompt_full_context(
if not isinstance(full_context_answers_dict, dict):
full_context_answers_dict = extract_and_parse(claude_answer_raw)
# Normalize 1:1 fields: convert single-element lists to strings
# Preserves legitimate multi-value lists (e.g., LOB, PROGRAM when passed to 1:1)
full_context_answers_dict = string_utils.normalize_one_to_one_answers_dict(
full_context_answers_dict
)
except Exception as e:
logging.error(f"Error processing high accuracy fields: {str(e)}")
full_context_answers_dict = {}
src/pipelines/saas/file_processing.py
+3 -8
View File
@@ -242,15 +242,10 @@ def run_one_to_one_prompts(
one_to_one_results, contract_text
)
################## Normalize before crosswalk ##################
# Normalize before get_crosswalk_fields (which expects/returns a list)
# get_crosswalk_fields and fill_na_mapping are mechanical steps (no LLM),
# so they won't introduce single-element lists that need normalization
one_to_one_results = string_utils.normalize_one_to_one_answers_dict(
one_to_one_results
)
################## Crosswalk Fields ##################
# All field format normalization is handled at prompt_calls level via field-aware parsers
# Derived fields (AARETE_DERIVED_*, NUM_*_SIGNATORY_*) are created as strings directly
# No additional normalization needed here
one_to_one_results = aarete_derived.get_crosswalk_fields(
[one_to_one_results], constants
)
src/pipelines/saas/prompts/prompt_calls.py
+140 -40
View File
@@ -1,10 +1,13 @@
import logging
from typing import Any
import src.config as config
import src.prompts.prompt_templates as prompt_templates
from src.constants.constants import Constants
from src.constants.investment_columns import FIELD_FORMAT_MAPPING
from src.prompts.fieldset import Field, FieldSet
from src.utils import llm_utils, string_utils
from src.utils.formatting_utils import normalize_field_value
from src.pipelines.shared.extraction import tin_npi_funcs
import json
@@ -19,8 +22,13 @@ def prompt_exhibit_level(
logging.debug(exhibit_level_fields.print_prompt_dict(constants))
if not exhibit_level_fields.contains_fields():
return {}
# Extract field names from FieldSet for format normalization
field_names = [field.field_name for field in exhibit_level_fields.fields]
prompt, _parser = prompt_templates.EXHIBIT_LEVEL(
exhibit_text, exhibit_level_fields.print_prompt_dict(constants)
exhibit_text,
exhibit_level_fields.print_prompt_dict(constants),
field_names=field_names,
)
llm_answer_raw = llm_utils.invoke_claude(
prompt,
@@ -34,7 +42,6 @@ def prompt_exhibit_level(
logging.debug(f"LLM raw output for {filename}: {llm_answer_raw}")
llm_answer_final = _parser(llm_answer_raw)
return llm_answer_final
@@ -93,6 +100,7 @@ def prompt_dynamic_primary(
)
logging.debug(f"Claude answer for {filename}; {field}: {llm_answer_raw}")
llm_answer_final = _parser(llm_answer_raw)
# Normalization is already done in the JSON parser
return llm_answer_final
@@ -124,11 +132,12 @@ def prompt_reimbursement_primary(
try:
llm_answer_final = _parser(llm_answer_raw)
# Normalization is already done in the JSON parser
return llm_answer_final
except ValueError as e:
logging.error(f"Error parsing LLM response: {e}")
logging.error(f"Raw LLM output: {llm_answer_raw}")
raise
return llm_answer_final
def prompt_methodology_breakout(
@@ -156,6 +165,17 @@ def prompt_methodology_breakout(
logging.debug(f"LLM Response for {filename}: {llm_response}")
try:
methodology_breakout_answers = _parser(llm_response)
# Normalize output - methodology_breakout returns list[dict] with methodology fields
# Use FieldSet to get field names from single source of truth
methodology_breakout_fields = FieldSet(
config.FIELD_JSON_PATH, field_type="methodology_breakout"
)
methodology_field_names = methodology_breakout_fields.list_fields()
# Handle both list and single dict cases
if isinstance(methodology_breakout_answers, dict):
methodology_breakout_answers = [methodology_breakout_answers]
# Normalization is already done in the JSON parser
return methodology_breakout_answers
except ValueError as e:
logging.error(f"Error parsing LLM response: {e}")
methodology_breakout_answers = []
@@ -189,6 +209,7 @@ def prompt_fee_schedule_breakout(
logging.debug(f"LLM Response for {filename}: {llm_answer_raw}")
try:
fs_breakout_dict = _parser(llm_answer_raw)
# Normalization is already done in the JSON parser
except ValueError as e:
fs_breakout_dict = {}
return fs_breakout_dict
@@ -219,6 +240,7 @@ def prompt_grouper_breakout(
logging.debug(f"LLM Response for {filename}: {llm_answer_raw}")
try:
grouper_breakout_dict = _parser(llm_answer_raw)
# Normalization is already done in the JSON parser
except:
grouper_breakout_dict = {}
@@ -248,7 +270,14 @@ def prompt_carveout_check(
)
carveout_answer = _parser(llm_answer_raw)
return carveout_answer[0]
# Parser normalizes to str format for CARVEOUT_CD, but returns as list
# Extract first element and ensure it's a string
if isinstance(carveout_answer, list) and len(carveout_answer) > 0:
result = carveout_answer[0]
return str(result) if result is not None else "N/A"
elif isinstance(carveout_answer, str):
return carveout_answer
return "N/A" # Default fallback
def prompt_special_case_breakout(breakout_template, term_answer, filename):
@@ -286,9 +315,12 @@ def prompt_special_case_breakout(breakout_template, term_answer, filename):
def prompt_lob_relationship(
answer_dict: dict, field: str, exhibit_text: str, filename: str
):
) -> str:
"""
Helper function to prompt LLM for LOB relationship based on the field and exhibit text.
Returns:
str: Normalized relationship string ("Inclusive" or "Exclusive")
"""
programs = answer_dict.get(field)
dynamic_primary_values = (
@@ -306,13 +338,15 @@ def prompt_lob_relationship(
instruction=prompt_templates.LOB_RELATIONSHIP_INSTRUCTION(),
)
llm_answer_final = _parser(llm_answer_raw)
# LOB_RELATIONSHIP returns a list (e.g., ["Inclusive"]), but we need a string
if isinstance(llm_answer_final, list) and len(llm_answer_final) > 0:
return str(llm_answer_final[0])
elif isinstance(llm_answer_final, str):
return llm_answer_final
else:
return "Exclusive" # Default fallback
# Parser normalizes to str format per field mapping, but may return list or str
# Normalize to ensure we always return a string
field_name = "LOB_PROGRAM_RELATIONSHIP" # Both fields use same format (str)
format_type = FIELD_FORMAT_MAPPING.get(field_name, "str")
# Normalize the result to str format (handles both list and str inputs)
normalized = normalize_field_value(field_name, llm_answer_final, format_type)
return normalized if normalized else "Exclusive"
def prompt_special_case_assignment(
@@ -353,9 +387,10 @@ def prompt_special_case_assignment(
)
index_answer = _parser(llm_answer_raw)
try:
# Parser returns a list, extract first element
# Parser returns a list of indices, normalize and extract first element
# This is a special case - we need the index as an integer
if isinstance(index_answer, list) and len(index_answer) > 0:
index_str = index_answer[0]
index_str = str(index_answer[0]) # Convert to string first
if not string_utils.is_empty(index_str):
return special_case_dicts[int(index_str)]
return {}
@@ -369,9 +404,11 @@ def prompt_full_context(
constants: Constants,
filename: str,
):
prompt_questions = full_context_fields.print_prompt_dict(constants)
# Extract field names for field-aware normalization
field_names = full_context_fields.list_fields()
full_context_prompt, _parser = prompt_templates.ONE_TO_ONE_MULTI_FIELD_TEMPLATE(
context=contract_text[
0 : min(
@@ -380,6 +417,7 @@ def prompt_full_context(
)
],
questions=prompt_questions,
field_names=field_names,
)
try:
@@ -393,11 +431,8 @@ def prompt_full_context(
)
full_context_answers_dict = _parser(claude_answer_raw)
# Normalize 1:1 fields: convert single-element lists to strings
# Preserves legitimate multi-value lists (e.g., LOB, PROGRAM when passed to 1:1)
full_context_answers_dict = string_utils.normalize_one_to_one_answers_dict(
full_context_answers_dict
)
# Field-aware normalization is already done in the JSON parser
# No additional normalization needed for fields in FIELD_FORMAT_MAPPING
except Exception as e:
logging.error(f"Error processing high accuracy fields: {str(e)}")
full_context_answers_dict = {}
@@ -457,7 +492,11 @@ def prompt_dynamic(text: str, field_prompts, filename):
Returns:
dict: A dictionary containing field names as keys and extracted answers as values.
"""
prompt, _parser = prompt_templates.EXHIBIT_LEVEL(text, field_prompts)
# Extract field names from field_prompts dict for format-aware normalization
field_names = list(field_prompts.keys()) if isinstance(field_prompts, dict) else []
prompt, _parser = prompt_templates.EXHIBIT_LEVEL(
text, field_prompts, field_names=field_names
)
logging.debug(f"Dynamic prompt for {filename}: {prompt}")
llm_answer_raw = llm_utils.invoke_claude(
prompt,
@@ -469,6 +508,7 @@ def prompt_dynamic(text: str, field_prompts, filename):
logging.debug(f"Dynamic answer for {filename}: {llm_answer_raw}")
llm_answer_final = _parser(llm_answer_raw)
# Normalization is already done in the JSON parser
return llm_answer_final
@@ -496,6 +536,15 @@ def prompt_exhibit_linkage(page_header, previous_header, filename):
usage_label="EXHIBIT_LINKAGE",
)
exhibit_is_different = _parser(claude_answer_raw)
# Normalize output - EXHIBIT_LINKAGE returns a list, normalize it
# No specific field name, so normalize based on content
if isinstance(exhibit_is_different, list):
# Extract first element if list (legacy behavior)
exhibit_is_different = (
exhibit_is_different[0]
if len(exhibit_is_different) > 0
else exhibit_is_different
)
return exhibit_is_different
@@ -525,7 +574,10 @@ def prompt_exhibit_header(page_content, filename):
usage_label="EXHIBIT_HEADER",
)
llm_answer_extracted = _parser(llm_answer_raw)
return llm_answer_extracted[0]
# Parser returns a list, extract first element
if isinstance(llm_answer_extracted, list) and len(llm_answer_extracted) > 0:
return llm_answer_extracted[0]
return llm_answer_extracted
def prompt_date_fix(date: str) -> str:
@@ -548,7 +600,12 @@ def prompt_date_fix(date: str) -> str:
instruction=prompt_templates.DATE_FIX_INSTRUCTION(),
usage_label="DATE_FIX",
)
return _parser(response)
parsed = _parser(response)
# Normalize output - date_fix returns a list, but we need a string
# No specific field name, so extract first element if list
if isinstance(parsed, list) and len(parsed) > 0:
return parsed[0]
return parsed
def prompt_derived_term_date(
@@ -577,7 +634,11 @@ def prompt_derived_term_date(
instruction=prompt_templates.DERIVED_TERM_DATE_INSTRUCTION(),
usage_label="DERIVED_TERM_DATE",
)
return _parser(response)
parsed = _parser(response)
# Parser returns a list, extract first element
if isinstance(parsed, list) and len(parsed) > 0:
return parsed[0]
return parsed
def prompt_dynamic_assignment(
@@ -624,6 +685,7 @@ def prompt_dynamic_assignment(
try:
llm_answer_final = _parser(llm_answer_raw)
# Normalization is already done in the JSON parser
return llm_answer_final
except:
logging.error(
@@ -704,14 +766,12 @@ def prompt_lesser_of_distribution(
logging.debug(
f"Applied lesser-of to '{service_term}' on page {page_num}: "
f"{reimb_term} → {llm_answer_final[:60]}..."
)
return (
llm_answer_final[0]
if isinstance(llm_answer_final, list)
else llm_answer_final
f"{reimb_term} → {llm_answer_final[:60] if isinstance(llm_answer_final, str) else str(llm_answer_final)[:60]}..."
)
# Parser returns a list, extract first element
if isinstance(llm_answer_final, list) and len(llm_answer_final) > 0:
return llm_answer_final[0]
return llm_answer_final
def prompt_lesser_of_check(
@@ -753,6 +813,7 @@ def prompt_lesser_of_check(
try:
llm_answer_final = _parser(llm_answer_raw)
# Normalization is already done in the JSON parser
logging.debug(f"LESSER_OF_CHECK extracted from pipes: {llm_answer_final}")
# Validate required fields
@@ -830,7 +891,12 @@ def prompt_exhibit_title_match(
try:
llm_answer_final = _parser(llm_answer_raw)
return llm_answer_final[0].strip().upper() # Return "YES" or "NO" as string
# Normalize output - EXHIBIT_TITLE_MATCH returns a list, but we need a string
# No specific field name, so extract first element and normalize
if isinstance(llm_answer_final, list) and len(llm_answer_final) > 0:
result = str(llm_answer_final[0]).strip().upper()
return result
return "NO" # Default to "NO" string
except:
return "NO" # Default to "NO" string
@@ -868,7 +934,12 @@ def provider_name_match_check(
# Extract last character (should be Y or N)
llm_answer_final = _parser(llm_answer_raw)
response_char = llm_answer_final[0]
# Normalize output - CHECK_PROVIDER_NAME_MATCH returns a list
# No specific field name, so extract first element
if isinstance(llm_answer_final, list) and len(llm_answer_final) > 0:
response_char = str(llm_answer_final[0])
else:
response_char = str(llm_answer_final) if llm_answer_final else "N"
if response_char == "Y":
logging.debug(
@@ -893,7 +964,14 @@ def provider_name_match_check(
return False
def prompt_split_reimb_dates(date_range: str, filename: str) -> dict:
def prompt_split_reimb_dates(date_range: str, filename: str) -> dict[str, str]:
"""
Split a date range string into effective and termination dates.
Returns:
dict: Dictionary with 'start_date' and 'end_date' keys, both normalized
to str format matching REIMB_EFFECTIVE_DT and REIMB_TERMINATION_DT.
"""
prompt, _parser = prompt_templates.SPLIT_REIMB_DATES(date_range)
# Invoke LLM to split the date range
@@ -909,7 +987,23 @@ 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
# Normalize date values to str format matching field format mapping
# The parser returns dict with 'start_date' and 'end_date' keys
# We normalize these values to match REIMB_EFFECTIVE_DT and REIMB_TERMINATION_DT formats
normalized_dict = {}
if isinstance(llm_answer_final, dict):
format_type = FIELD_FORMAT_MAPPING.get("REIMB_EFFECTIVE_DT", "str")
if "start_date" in llm_answer_final:
normalized_dict["start_date"] = normalize_field_value(
"REIMB_EFFECTIVE_DT", llm_answer_final["start_date"], format_type
)
if "end_date" in llm_answer_final:
normalized_dict["end_date"] = normalize_field_value(
"REIMB_TERMINATION_DT", llm_answer_final["end_date"], format_type
)
return normalized_dict if normalized_dict else llm_answer_final
def prompt_provider_info(
@@ -957,11 +1051,17 @@ def prompt_provider_info(
logging.debug(f"Parsed provider info on page {page_num}: {providers}")
# Ensure we have a list of providers (not a dict or other type)
if not isinstance(providers, list):
# If not a list or dict, create a default list with error info
logging.warning(
f"Warning: Unexpected format in provider info response: {type(providers)}"
)
providers = [{"TIN": "N/A", "NPI": "N/A", "NAME": "PARSING_ERROR"}]
# If single provider returned as dict, wrap in list
if isinstance(providers, dict):
providers = [providers]
else:
# If not a list or dict, create a default list with error info
logging.warning(
f"Warning: Unexpected format in provider info response: {type(providers)}"
)
providers = [{"TIN": "N/A", "NPI": "N/A", "NAME": "PARSING_ERROR"}]
# Normalization is already done in the JSON parser
# VALIDATION LAYER - Cross-check with regex findings
page_text = text_dict[page_num]
src/pipelines/shared/extraction/one_to_n_funcs.py
+4 -28
View File
@@ -680,6 +680,7 @@ def split_reimb_dates(one_to_n_results: list, filename: str) -> list:
date_range, filename
)
# Update the specific record with parsed dates
# Values are already normalized to str format in prompt_calls.py
if llm_answer_final and isinstance(llm_answer_final, dict):
if "start_date" in llm_answer_final:
record["REIMB_EFFECTIVE_DT"] = llm_answer_final["start_date"]
@@ -782,34 +783,9 @@ def one_to_n_cleaning(
all_exhibit_rows = split_reimb_dates(all_exhibit_rows, filename)
################################ Normalize Single-Value Fields to Strings ################################
# REIMB_TERM, SERVICE_TERM, and LOB_PROGRAM_RELATIONSHIP should always be strings (not lists)
for answer_dict in all_exhibit_rows:
# LOB_PROGRAM_RELATIONSHIP should be a string (not a list)
if "LOB_PROGRAM_RELATIONSHIP" in answer_dict:
lob_program_rel = answer_dict["LOB_PROGRAM_RELATIONSHIP"]
if isinstance(lob_program_rel, list):
# If list, take first element (shouldn't happen, but defensive)
answer_dict["LOB_PROGRAM_RELATIONSHIP"] = (
str(lob_program_rel[0]) if lob_program_rel else "Exclusive"
)
elif lob_program_rel is not None:
answer_dict["LOB_PROGRAM_RELATIONSHIP"] = str(lob_program_rel)
else:
answer_dict["LOB_PROGRAM_RELATIONSHIP"] = "Exclusive"
# LOB_PRODUCT_RELATIONSHIP should be a string (not a list)
if "LOB_PRODUCT_RELATIONSHIP" in answer_dict:
lob_product_rel = answer_dict["LOB_PRODUCT_RELATIONSHIP"]
if isinstance(lob_product_rel, list):
# If list, take first element (shouldn't happen, but defensive)
answer_dict["LOB_PRODUCT_RELATIONSHIP"] = (
str(lob_product_rel[0]) if lob_product_rel else "Exclusive"
)
elif lob_product_rel is not None:
answer_dict["LOB_PRODUCT_RELATIONSHIP"] = str(lob_product_rel)
else:
answer_dict["LOB_PRODUCT_RELATIONSHIP"] = "Exclusive"
# REIMB_TERM, SERVICE_TERM should always be strings (not lists)
# LOB_PROGRAM_RELATIONSHIP and LOB_PRODUCT_RELATIONSHIP are already normalized to strings
# in prompt_calls.prompt_lob_relationship(), so no additional normalization needed here
return all_exhibit_rows
src/pipelines/shared/extraction/one_to_one_funcs.py
+12 -23
View File
@@ -197,29 +197,22 @@ def run_full_context_fields(
)
# Normalize PAYER_STATE to two-letter abbreviation
if "PAYER_STATE" in full_context_answers_dict:
full_context_answers_dict["PAYER_STATE"] = (
string_utils.normalize_state_to_abbreviation(
full_context_answers_dict["PAYER_STATE"]
)
)
# normalize_state_field correctly handles list[str] format from field-aware parser
full_context_answers_dict = string_utils.normalize_state_field(
full_context_answers_dict, field_name="PAYER_STATE"
)
# normalize PROVIDER_STATE to two-letter abbreviation
if "PROVIDER_STATE" in full_context_answers_dict:
full_context_answers_dict = string_utils.normalize_state_field(
full_context_answers_dict, field_name="PROVIDER_STATE"
)
# Normalize PROVIDER_STATE to two-letter abbreviation
# normalize_state_field correctly handles list[str] format from field-aware parser
full_context_answers_dict = string_utils.normalize_state_field(
full_context_answers_dict, field_name="PROVIDER_STATE"
)
# Run Special Case Breakout on any breakout terms
full_context_answers_dict = one_to_one_breakout(full_context_answers_dict, filename)
# Normalize breakout fields: prompt_special_case_breakout uses _json_dict_parser
# which may return lists, so normalize any newly added fields
full_context_answers_dict = string_utils.normalize_one_to_one_answers_dict(
full_context_answers_dict
)
# Field-aware normalization is already done in prompt_calls.prompt_full_context()
# and prompt_special_case_breakout() uses field-aware parsers
# No additional normalization needed for fields in FIELD_FORMAT_MAPPING
# Handle separate one-to-one fields that require individual processing
separate_fields = one_to_one_fields.filter(field_type="full_context_separate")
@@ -309,14 +302,10 @@ def one_to_one_breakout(
breakout_answer_dict = prompt_calls.prompt_special_case_breakout(
breakout_template, term_answer, filename
)
# Normalize breakout fields for 1:1: _json_dict_parser may return lists for values
# Convert single-element lists to strings, preserve multi-value lists
# Note: This normalization is only for 1:1 fields; 1:N breakouts are handled separately
# Field-aware normalization is already done in prompt_special_case_breakout()
# which uses field-aware parsers based on the breakout template's field_names
if isinstance(breakout_answer_dict, dict):
breakout_answer_dict = string_utils.normalize_one_to_one_answers_dict(
breakout_answer_dict
)
full_context_answers_dict.update(breakout_answer_dict)
full_context_answers_dict.update(breakout_answer_dict)
return full_context_answers_dict
src/pipelines/shared/postprocessing/aarete_derived.py
+10 -17
View File
@@ -2,9 +2,11 @@ import logging
import src.config as config
import src.utils.string_utils as string_utils
from src.constants.constants import Constants
from src.constants.investment_columns import FIELD_FORMAT_MAPPING
from src.crosswalk.crosswalk_builder import CrosswalkBuilder
from src.prompts.fieldset import FieldSet
from src.utils.crosswalk_utils import apply_crosswalk
from src.utils.formatting_utils import normalize_field_value
def fill_na_from_field(results_dict, to_field, from_field, crosswalk_path):
@@ -172,16 +174,9 @@ def get_crosswalk_fields(answer_dicts: list, constants: Constants):
"""
Apply crosswalk mappings to derive target fields from source fields.
Fields that should be single values (strings) even when crosswalk creates lists:
- AARETE_DERIVED_CLAIM_TYPE_CD: Claim type is a single classification value
- Other AARETE_DERIVED_* fields that represent single classifications (not multi-value like LOB)
Normalizes output based on FIELD_FORMAT_MAPPING to ensure correct format
(str vs list[str]) for each target field.
"""
# Fields that should always be single values (strings), not lists
# These are typically classification/derived fields that represent a single value
SINGLE_VALUE_FIELDS = {
"AARETE_DERIVED_CLAIM_TYPE_CD",
# Add other single-value derived fields here as needed
}
crosswalk_fields = FieldSet(file_path=config.FIELD_JSON_PATH, crosswalk=True)
for to_field in crosswalk_fields.fields:
@@ -233,14 +228,12 @@ def get_crosswalk_fields(answer_dicts: list, constants: Constants):
individual_from_field_value
)
# Assign the result: convert to string if it's a single-value field with one element
if (
to_field_name in SINGLE_VALUE_FIELDS
and len(to_field_answer_list) == 1
):
answer_dict[to_field_name] = to_field_answer_list[0]
else:
answer_dict[to_field_name] = to_field_answer_list
# Normalize the result based on FIELD_FORMAT_MAPPING
format_type = FIELD_FORMAT_MAPPING.get(to_field_name, "list[str]")
normalized_value = normalize_field_value(
to_field_name, to_field_answer_list, format_type
)
answer_dict[to_field_name] = normalized_value
# DO NOT update from_field_name - preserve its original format
# The crosswalk should only create/update the target field (to_field_name),
src/pipelines/shared/postprocessing/postprocess.py
+2 -2
View File
@@ -1,6 +1,6 @@
import src.config as config
from src.constants.constants import Constants
from src.constants.investment_columns import COLUMN_ORDER
from src.constants.investment_columns import FIELD_FORMAT_MAPPING
from src.pipelines.shared.postprocessing import postprocessing_funcs
@@ -88,6 +88,6 @@ def postprocess(df, constants: Constants):
df = postprocessing_funcs.attach_sid_column(df)
# Standardize output column order - this should ALWAYS be the final postprocessing step
df = postprocessing_funcs.reorder_columns(df, COLUMN_ORDER)
df = postprocessing_funcs.reorder_columns(df, FIELD_FORMAT_MAPPING)
return df
src/pipelines/shared/postprocessing/postprocessing_funcs.py
+6 -1
View File
@@ -311,7 +311,9 @@ def generate_reimb_ids(df: pd.DataFrame) -> pd.DataFrame:
return df_temp
def reorder_columns(df: pd.DataFrame, column_order: list[str]) -> pd.DataFrame:
def reorder_columns(
df: pd.DataFrame, field_format_mapping: dict[str, str]
) -> pd.DataFrame:
"""
Reorders the columns of the DataFrame based on the given column order.
Steps:
@@ -326,6 +328,9 @@ def reorder_columns(df: pd.DataFrame, column_order: list[str]) -> pd.DataFrame:
Returns:
pd.DataFrame: The reordered DataFrame.
"""
# Get a list of column names, preserving order
column_order = list(field_format_mapping.keys())
# Create a copy to avoid fragmentation
df_copy = df.copy()
src/pipelines/shared/preprocessing/hybrid_smart_chunking_funcs.py
+12 -9
View File
@@ -284,10 +284,6 @@ def run_hybrid_smart_chunked_fields(
answers_dict, field_name="PAYER_STATE"
)
# Normalize all 1:1 fields: convert single-element lists to strings
# Preserves legitimate multi-value lists (e.g., LOB, PROGRAM when passed to 1:1)
answers_dict = string_utils.normalize_one_to_one_answers_dict(answers_dict)
return answers_dict
except Exception as e:
@@ -372,7 +368,7 @@ def prompt_hsc_single_field(
constants
) # Returns dict of {field_name : prompt}
prompt, _parser = prompt_templates.ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(
context, llm_prompt
context, llm_prompt, field_name=field.field_name
)
logging.debug(f"Field: {field.field_name}")
@@ -388,12 +384,19 @@ def prompt_hsc_single_field(
)
try:
llm_answer_final = _parser(llm_answer_raw) # returns dict
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":
# Check for N/A - handle both string and list formats
field_value_str = (
field_value
if isinstance(field_value, str)
else (
field_value[0]
if isinstance(field_value, list) and field_value
else "N/A"
)
)
if field_value_str == "N/A":
field.field_type = "full_context"
if field.field_name == "PAYER_NAME":
field.prompt = (
src/prompts/prompt_templates.py
+191 -50
View File
@@ -32,15 +32,43 @@ JSON_LIST_OF_DICTS_FORMAT_INSTRUCTIONS = """Return your final answer as a valid
# Parser helper functions for prompt templates
def _json_dict_parser(raw_output: str) -> dict:
"""Parse JSON dictionary from LLM output."""
"""Parse JSON dictionary from LLM output (without field normalization).
For field-aware parsing, use _create_json_dict_parser() instead.
"""
return json_utils.parse_json_dict(raw_output)
def _json_list_parser(raw_output: str) -> list:
"""Parse JSON list from LLM output."""
"""Parse JSON list from LLM output (without field normalization).
For field-aware parsing, use _create_json_list_parser() instead.
"""
return json_utils.parse_json_list(raw_output)
def _create_json_dict_parser(field_names: list[str] | None = None):
"""Create a JSON dict parser with field names bound for normalization."""
def parser(raw_output: str) -> dict:
return json_utils.parse_json_dict(raw_output, field_names=field_names)
return parser
def _create_json_list_parser(
field_name: str | None = None, expected_format: str | None = None
):
"""Create a JSON list parser with field name bound for normalization."""
def parser(raw_output: str) -> list:
return json_utils.parse_json_list(
raw_output, field_name=field_name, expected_format=expected_format
)
return parser
# Module-level cached Constants instance for instruction functions
# This is lazily initialized on first use and reused across all instruction calls
_cached_constants = None
@@ -235,10 +263,17 @@ Briefly explain your answer, then put the final answer in the properly-formatted
{JSON_DICT_WITH_LISTS_FORMAT_INSTRUCTIONS}"""
def EXHIBIT_LEVEL(context, fields) -> Tuple[str, Callable[[str], dict]]:
def EXHIBIT_LEVEL(
context, fields, field_names: list[str] | None = None
) -> Tuple[str, Callable[[str], dict]]:
"""Returns ONLY dynamic content for exhibit level extraction.
Call EXHIBIT_LEVEL_INSTRUCTION() separately for the cached instruction.
Args:
context: The exhibit text to analyze.
fields: Formatted string of field definitions.
field_names: Optional list of field names for format-aware normalization.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
@@ -250,7 +285,8 @@ Here are the attributes to be included in the dictionary, and instructions on ho
Here is the text to analyze:
{context.replace('"', "'")}"""
return (prompt, _json_dict_parser)
parser = _create_json_dict_parser(field_names) if field_names else _json_dict_parser
return (prompt, parser)
def DYNAMIC_PRIMARY_INSTRUCTION() -> str:
@@ -277,6 +313,11 @@ def DYNAMIC_PRIMARY(
"""Returns ONLY dynamic content for text-based dynamic primary extraction.
Call DYNAMIC_PRIMARY_TEXT_INSTRUCTION() separately for the cached instruction.
Args:
context: The exhibit text to analyze.
field_name: The field name being extracted (used for format-aware normalization).
field_prompt: The prompt/description for the field.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
"""
@@ -287,7 +328,9 @@ def DYNAMIC_PRIMARY(
Here is the text to analyze:
{context.replace('"', "'")}"""
return (prompt, _json_list_parser)
# Create parser with field_name bound for format-aware normalization
parser = _create_json_list_parser(field_name=field_name)
return (prompt, parser)
def REIMB_DATES_ASSIGNMENT_INSTRUCTION() -> str:
@@ -370,6 +413,13 @@ def REIMB_DATES_ASSIGNMENT(
"""Returns ONLY dynamic content for REIMB_DATES assignment.
Call REIMB_DATES_ASSIGNMENT_INSTRUCTION() separately for the cached instruction.
Args:
service_term: Service description
reimb_term: Reimbursement term text
field_prompt: Prompt/definition for REIMB_DATES field
exhibit_text_simplified: Simplified exhibit text
page_num: Page number
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
@@ -387,7 +437,9 @@ This is the term you need to find the date range for. It appears on page {page_n
Service Term: "{service_term}"
Reimbursement Term: "{reimb_term}" """
return (prompt, _json_dict_parser)
# Create parser with REIMB_DATES field name bound for format-aware normalization
parser = _create_json_dict_parser(field_names=["REIMB_DATES"])
return (prompt, parser)
def DYNAMIC_ASSIGNMENT_INSTRUCTION() -> str:
@@ -460,6 +512,14 @@ def DYNAMIC_ASSIGNMENT(
Returns prompt for dynamic assignment extraction.
Call DYNAMIC_ASSIGNMENT_INSTRUCTION() separately for the cached instruction.
Args:
service_term: Service description
reimb_term: Reimbursement term text
field_name: Name of the field being assigned (used for format-aware normalization)
field_prompt: Prompt/definition for the field
exhibit_text_simplified: Simplified exhibit text
page_num: Page number
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
@@ -478,7 +538,10 @@ Reimbursement Term: "{reimb_term}"
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)
# Create parser with field_name bound for format-aware normalization
# The dict will contain {field_name: value}, so we need to normalize that field
parser = _create_json_dict_parser(field_names=[field_name])
return (prompt, parser)
def REIMBURSEMENT_PRIMARY(context) -> Tuple[str, Callable[[str], list]]:
@@ -497,7 +560,11 @@ 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)
# Use helper function to create parser for list[dict[str,str]]
parser = _create_json_list_parser(
field_name=None, expected_format="list[dict[str,str]]"
)
return (prompt, parser)
def REIMBURSEMENT_PRIMARY_INSTRUCTION() -> str:
@@ -827,7 +894,11 @@ 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)
# Use helper function to create parser for list[dict[str,str]]
parser = _create_json_list_parser(
field_name=None, expected_format="list[dict[str,str]]"
)
return (prompt, parser)
def METHODOLOGY_BREAKOUT_INSTRUCTION() -> str:
@@ -907,7 +978,13 @@ def FEE_SCHEDULE_BREAKOUT(
Analyze and respond to the following text, specifically for {fee_schedule} Fee Schedule:
Methodology: {methodology.replace('"', "'")}"""
return (prompt, _json_dict_parser)
# Extract field names for format-aware normalization
fee_schedule_fields = FieldSet(
config.FIELD_JSON_PATH, field_type="fee_schedule_breakout"
)
field_names = [field.field_name for field in fee_schedule_fields.fields]
parser = _create_json_dict_parser(field_names=field_names)
return (prompt, parser)
def GROUPER_BREAKOUT_INSTRUCTION() -> str:
@@ -949,13 +1026,25 @@ Here are the service and reimbursement terms to analyze:
SERVICE: {service}
REIMBURSEMENT TERM: {term}"""
return (prompt, _json_dict_parser)
# Extract field names for format-aware normalization
grouper_fields = FieldSet(config.FIELD_JSON_PATH, field_type="grouper_breakout")
field_names = [field.field_name for field in grouper_fields.fields]
parser = _create_json_dict_parser(field_names=field_names)
return (prompt, parser)
def SPECIAL_CASE_BREAKOUT(term, questions) -> Tuple[str, Callable[[str], dict]]:
def SPECIAL_CASE_BREAKOUT(
term, questions, field_names: list[str] | None = None
) -> Tuple[str, Callable[[str], dict]]:
"""
Generic prompt for special case breakout fields when no additional descriptions are needed.
Args:
term: The term text to analyze
questions: Formatted string of field definitions
field_names: Optional list of field names for format-aware normalization.
If None, will attempt to extract from questions string.
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
@@ -974,7 +1063,9 @@ Here is the term to analyze:
Briefly explain your answer, then put your final answer in JSON dictionary format.
"""
return (prompt, _json_dict_parser)
# Create parser with field_names bound for format-aware normalization
parser = _create_json_dict_parser(field_names=field_names)
return (prompt, parser)
def RATE_ESCALATOR_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
@@ -982,16 +1073,17 @@ def RATE_ESCALATOR_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
Creates a prompt to extract rate escalator field information from a rate escalator statement.
Args:
rate_escalator_statement: The text containing rate escalator information
rate_escalator_questions: Dictionary or list of questions with instructions for each field
term: The term text to analyze
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
rate_escalator_fields = FieldSet(
config.FIELD_JSON_PATH, field_type="rate_escalator_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
)
questions = rate_escalator_fields.print_prompt_dict()
field_names = [field.field_name for field in rate_escalator_fields.fields]
return SPECIAL_CASE_BREAKOUT(term, questions, field_names=field_names)
def TRIGGER_CAP_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
@@ -1001,10 +1093,12 @@ def TRIGGER_CAP_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
trigger_cap_fields = FieldSet(
config.FIELD_JSON_PATH, field_type="trigger_cap_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
)
questions = trigger_cap_fields.print_prompt_dict()
field_names = [field.field_name for field in trigger_cap_fields.fields]
return SPECIAL_CASE_BREAKOUT(term, questions, field_names=field_names)
def OUTLIER_BREAKOUT_INSTRUCTION() -> str:
@@ -1046,20 +1140,30 @@ def OUTLIER_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
Here is the text to analyze and respond to:
{term}"""
return (prompt, _json_dict_parser)
# Extract field names for format-aware normalization
outlier_fields = FieldSet(config.FIELD_JSON_PATH, field_type="outlier_breakout")
field_names = [field.field_name for field in outlier_fields.fields]
parser = _create_json_dict_parser(field_names=field_names)
return (prompt, parser)
def FACILITY_ADJUSTMENT_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for facility adjustment breakout.
Args:
term: The facility adjustment term text to analyze
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
facility_adjustment_fields = FieldSet(
config.FIELD_JSON_PATH, field_type="facility_adjustment_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
)
questions = facility_adjustment_fields.print_prompt_dict()
# Extract field names for format-aware normalization
field_names = [field.field_name for field in facility_adjustment_fields.fields]
return SPECIAL_CASE_BREAKOUT(term, questions, field_names=field_names)
def SEQUESTRATION_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
@@ -1069,10 +1173,12 @@ def SEQUESTRATION_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
sequestration_fields = FieldSet(
config.FIELD_JSON_PATH, field_type="sequestration_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
)
questions = sequestration_fields.print_prompt_dict()
field_names = [field.field_name for field in sequestration_fields.fields]
return SPECIAL_CASE_BREAKOUT(term, questions, field_names=field_names)
def DISCOUNT_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
@@ -1082,10 +1188,10 @@ def DISCOUNT_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="discount_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
discount_fields = FieldSet(config.FIELD_JSON_PATH, field_type="discount_breakout")
questions = discount_fields.print_prompt_dict()
field_names = [field.field_name for field in discount_fields.fields]
return SPECIAL_CASE_BREAKOUT(term, questions, field_names=field_names)
def PREMIUM_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
@@ -1095,10 +1201,10 @@ def PREMIUM_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="premium_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
premium_fields = FieldSet(config.FIELD_JSON_PATH, field_type="premium_breakout")
questions = premium_fields.print_prompt_dict()
field_names = [field.field_name for field in premium_fields.fields]
return SPECIAL_CASE_BREAKOUT(term, questions, field_names=field_names)
def ADDITION_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
@@ -1108,10 +1214,10 @@ def ADDITION_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="addition_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
addition_fields = FieldSet(config.FIELD_JSON_PATH, field_type="addition_breakout")
questions = addition_fields.print_prompt_dict()
field_names = [field.field_name for field in addition_fields.fields]
return SPECIAL_CASE_BREAKOUT(term, questions, field_names=field_names)
def STOP_LOSS_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
@@ -1121,10 +1227,10 @@ def STOP_LOSS_BREAKOUT(term) -> Tuple[str, Callable[[str], dict]]:
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
questions = FieldSet(
config.FIELD_JSON_PATH, field_type="stop_loss_breakout"
).print_prompt_dict()
return SPECIAL_CASE_BREAKOUT(term, questions)
stop_loss_fields = FieldSet(config.FIELD_JSON_PATH, field_type="stop_loss_breakout")
questions = stop_loss_fields.print_prompt_dict()
field_names = [field.field_name for field in stop_loss_fields.fields]
return SPECIAL_CASE_BREAKOUT(term, questions, field_names=field_names)
#####################################################################################
@@ -1397,11 +1503,16 @@ Here are the descriptions to choose from:
def ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(
context: str, field_prompt: dict[str, str]
context: str, field_prompt: dict[str, str], field_name: str | None = None
) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for one-to-one single field extraction.
Args:
context: Contract text context
field_prompt: Dictionary mapping field name to prompt question
field_name: Optional field name for field-aware normalization
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
@@ -1415,7 +1526,12 @@ 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)
# Use field-aware parser if field_name is provided
if field_name:
parser = _create_json_dict_parser(field_names=[field_name])
else:
parser = _json_dict_parser
return (prompt, parser)
def ONE_TO_ONE_SINGLE_FIELD_INSTRUCTION() -> str:
@@ -1596,11 +1712,16 @@ Note: PROVIDER NAME B is typically a single name but may contain multiple names.
def ONE_TO_ONE_MULTI_FIELD_TEMPLATE(
context, questions: dict[str, str]
context, questions: dict[str, str], field_names: list[str] | None = None
) -> Tuple[str, Callable[[str], dict]]:
"""
Returns prompt and parser for one-to-one multi-field extraction.
Args:
context: Contract text context
questions: Dictionary mapping field names to prompt questions
field_names: Optional list of field names for field-aware normalization
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
"""
@@ -1631,7 +1752,12 @@ The term length appears in two places but I'm selecting...
"term_length": "36 months"
}}
"""
return (prompt, _json_dict_parser)
# Use field-aware parser if field_names provided
if field_names:
parser = _create_json_dict_parser(field_names=field_names)
else:
parser = _json_dict_parser
return (prompt, parser)
def ONE_TO_ONE_MULTI_FIELD_INSTRUCTION() -> str:
@@ -1988,13 +2114,16 @@ def CARVEOUT_CHECK(
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
The parser normalizes the output for CARVEOUT_CD field format (str).
"""
prompt = f"""[REIMBURSEMENT TERM]
Here is the Reimbursement Term to analyze:
Service: {service_term}
Reimbursement Term: {reimb_term.replace('"', "'")}"""
return (prompt, _json_list_parser)
# Use field-aware parser to normalize CARVEOUT_CD (str format)
parser = _create_json_list_parser(field_name="CARVEOUT_CD")
return (prompt, parser)
def EFFECTIVE_DATE_FIX_PROMPT() -> Tuple[str, Callable[[str], dict]]:
@@ -2066,10 +2195,15 @@ def SPLIT_REIMB_DATES(date_range: str) -> Tuple[str, Callable[[str], dict]]:
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON dict output.
Note: The output dict contains 'start_date' and 'end_date' keys, which are
normalized manually in split_reimb_dates() to match REIMB_EFFECTIVE_DT and
REIMB_TERMINATION_DT field formats (both str).
"""
prompt = f"""[DATE RANGE TO EXTRACT]
{date_range}"""
# Use standard parser - normalization happens manually in split_reimb_dates()
# because the output keys are 'start_date'/'end_date', not the field names
return (prompt, _json_dict_parser)
@@ -2110,6 +2244,9 @@ def LOB_RELATIONSHIP(
Returns:
Tuple of (prompt_string, parser_function) where parser expects JSON list output.
The parser normalizes the output for LOB_PROGRAM_RELATIONSHIP or
LOB_PRODUCT_RELATIONSHIP field format (str). Note: This parser is used for
both fields, so normalization is based on str format.
"""
prompt = f"""[LOB AND PROGRAMS IDENTIFIED]
It was previously identified that the Exhibit contains the following LOB and Programs:
@@ -2119,7 +2256,11 @@ It was previously identified that the Exhibit contains the following LOB and Pro
Here is the Exhibit to be analyzed:
{exhibit_text.replace('"', "'")}"""
return (prompt, _json_list_parser)
# Use field-aware parser to normalize relationship field (str format)
# Note: This is used for both LOB_PROGRAM_RELATIONSHIP and LOB_PRODUCT_RELATIONSHIP,
# both of which are str format, so we can use either field name
parser = _create_json_list_parser(field_name="LOB_PROGRAM_RELATIONSHIP")
return (prompt, parser)
def DERIVED_TERM_DATE_INSTRUCTION() -> str:
src/testbed/add_reimb_id_to_testbed.py
+2 -2
View File
@@ -9,7 +9,7 @@ Run it with `poetry run python -m src.testbed.add_reimb_id_to_testbed`
# Imports
import pandas as pd
from src.constants.investment_columns import COLUMN_ORDER
from src.constants.investment_columns import FIELD_FORMAT_MAPPING
from src.pipelines.shared.postprocessing.postprocessing_funcs import (
generate_reimb_ids,
reorder_columns,
@@ -19,7 +19,7 @@ from src.pipelines.shared.postprocessing.postprocessing_funcs import (
testbed = pd.read_excel("Doczy-Testbed.xlsx")
# Generate reimbursement IDs and reorder columns
testbed_w_ids = reorder_columns(generate_reimb_ids(testbed), COLUMN_ORDER)
testbed_w_ids = reorder_columns(generate_reimb_ids(testbed), FIELD_FORMAT_MAPPING)
# Save the updated testbed with reimbursement IDs
testbed_w_ids.to_excel("Doczy-Testbed-with-Reimb-IDs.xlsx", index=False)
src/testbed/testbed_utils.py
+2 -2
View File
@@ -5,7 +5,7 @@ from collections import Counter
import pandas as pd
import src.config as config
import src.utils.string_utils as string_utils
from src.constants.investment_columns import COLUMN_ORDER
from src.constants.investment_columns import FIELD_FORMAT_MAPPING
from rapidfuzz import fuzz
# Imports for the testbed postprocessing functions
@@ -941,7 +941,7 @@ def testbed_postprocess(df):
df = postprocessing_funcs.generate_reimb_ids(df)
# Standardize output column order - this should ALWAYS be the final postprocessing step
df = postprocessing_funcs.reorder_columns(df, COLUMN_ORDER)
df = postprocessing_funcs.reorder_columns(df, FIELD_FORMAT_MAPPING)
return df
src/tests/test_json_parsers.py
+353 -182
View File
@@ -2,14 +2,15 @@
Unit tests for JSON utils module.
Tests the new JSON parsing functions that replace pipe-delimited output format.
Also tests field-aware normalization functionality.
"""
import pytest
from src.utils.json_utils import (
parse_json_dict,
parse_json_list,
parse_json_dict_or_list,
)
from src.constants.investment_columns import FIELD_FORMAT_MAPPING
class TestParseJsonDict:
@@ -98,34 +99,129 @@ class TestParseJsonDict:
result = parse_json_dict(output)
assert result == {"field": "café", "emoji": "😀"}
def test_no_dict_raises_error(self):
"""Test that ValueError is raised when no dict is found."""
output = "This is just text without any JSON"
with pytest.raises(ValueError, match="No valid JSON dictionary found"):
parse_json_dict(output)
def test_only_list_raises_error(self):
"""Test that ValueError is raised when only a list is found."""
output = '["item1", "item2"]'
with pytest.raises(ValueError, match="No valid JSON dictionary found"):
parse_json_dict(output)
def test_malformed_json_raises_error(self):
"""Test that ValueError is raised for malformed JSON."""
output = '{"field": "value"' # Missing closing brace
with pytest.raises(ValueError, match="No valid JSON dictionary found"):
parse_json_dict(output)
def test_non_string_input_raises_error(self):
"""Test that TypeError is raised for non-string input."""
with pytest.raises(TypeError, match="Expected string input"):
parse_json_dict(123)
def test_dict_with_list_values(self):
"""Test parsing dict with list values."""
output = '{"items": ["a", "b", "c"], "counts": [1, 2, 3]}'
output = '{"items": ["a", "b", "c"], "numbers": [1, 2, 3]}'
result = parse_json_dict(output)
assert result == {"items": ["a", "b", "c"], "counts": [1, 2, 3]}
# Lists are preserved, but if field is in mapping, will be normalized
assert result["items"] == ["a", "b", "c"]
assert result["numbers"] == [1, 2, 3]
def test_dict_with_field_names_normalization(self):
"""Test parsing dict with field names for normalization."""
# PAYER_STATE should be normalized to list[str]
output = '{"PAYER_STATE": "IL", "CONTRACT_TITLE": "Test Contract"}'
result = parse_json_dict(output, field_names=["PAYER_STATE", "CONTRACT_TITLE"])
assert result["PAYER_STATE"] == ["IL"] # Normalized to list
assert result["CONTRACT_TITLE"] == "Test Contract" # Stays as str
def test_dict_with_field_names_list_to_str(self):
"""Test normalization of list to str for single-value fields."""
# CONTRACT_TITLE should be normalized to str (single element list -> str)
output = '{"CONTRACT_TITLE": ["Test Contract"]}'
result = parse_json_dict(output, field_names=["CONTRACT_TITLE"])
assert result["CONTRACT_TITLE"] == "Test Contract"
def test_dict_with_field_names_str_to_list(self):
"""Test normalization of str to list for list fields."""
# PAYER_STATE should be normalized to list[str]
output = '{"PAYER_STATE": "IL", "PROVIDER_STATE": "CA"}'
result = parse_json_dict(output, field_names=["PAYER_STATE", "PROVIDER_STATE"])
assert result["PAYER_STATE"] == ["IL"]
assert result["PROVIDER_STATE"] == ["CA"]
def test_dict_with_field_names_multiple_list_fields(self):
"""Test normalization of multiple list fields."""
output = '{"PAYER_STATE": "IL", "PROVIDER_STATE": "CA", "LOB": ["Medicare"]}'
result = parse_json_dict(
output, field_names=["PAYER_STATE", "PROVIDER_STATE", "LOB"]
)
assert result["PAYER_STATE"] == ["IL"]
assert result["PROVIDER_STATE"] == ["CA"]
assert result["LOB"] == ["Medicare"] # Already a list, stays as list
def test_dict_with_field_names_prov_info_json(self):
"""Test normalization of PROV_INFO_JSON field."""
# PROV_INFO_JSON should be list[dict[str,str]]
output = (
'{"PROV_INFO_JSON": [{"NAME": "Provider", "TIN": "123", "NPI": "456"}]}'
)
result = parse_json_dict(output, field_names=["PROV_INFO_JSON"])
assert isinstance(result["PROV_INFO_JSON"], list)
assert len(result["PROV_INFO_JSON"]) == 1
assert result["PROV_INFO_JSON"][0] == {
"NAME": "Provider",
"TIN": "123",
"NPI": "456",
}
def test_dict_with_field_names_partial_normalization(self):
"""Test that only specified fields are normalized when field_names provided."""
output = '{"PAYER_STATE": "IL", "CONTRACT_TITLE": ["Test"], "UNKNOWN_FIELD": "value"}'
result = parse_json_dict(output, field_names=["PAYER_STATE"])
assert result["PAYER_STATE"] == [
"IL"
] # Normalized (in field_names and mapping)
# CONTRACT_TITLE not in field_names, so not normalized even though it's in mapping
assert result["CONTRACT_TITLE"] == ["Test"] # Preserved as-is
# UNKNOWN_FIELD not in field_names and not in mapping, preserved as-is
assert result["UNKNOWN_FIELD"] == "value" # Preserved as-is
def test_dict_without_field_names_still_normalizes(self):
"""Test that dict normalization happens even without field_names."""
# When field_names is None, all fields in the dict are normalized if they're in mapping
output = '{"PAYER_STATE": "IL", "CONTRACT_TITLE": ["Test"]}'
result = parse_json_dict(output) # No field_names provided
# Should normalize based on field format mapping
assert result["PAYER_STATE"] == ["IL"] # Normalized based on mapping
assert result["CONTRACT_TITLE"] == "Test" # Normalized (list -> str)
def test_dict_with_unknown_field_no_normalization(self):
"""Test that unknown fields are not normalized."""
output = '{"UNKNOWN_FIELD": "value"}'
result = parse_json_dict(output, field_names=["UNKNOWN_FIELD"])
# Unknown field should remain as-is (no format mapping)
assert result["UNKNOWN_FIELD"] == "value"
def test_dict_field_aware_verification(self):
"""Comprehensive test to verify dict parser is truly field-aware."""
# Test with multiple fields of different types
output = '{"PAYER_STATE": "IL", "CONTRACT_TITLE": ["My Contract"], "LOB": "Medicare", "PROV_INFO_JSON": [{"NAME": "Provider"}]}'
# Test with field_names - only specified fields should be normalized
result = parse_json_dict(output, field_names=["PAYER_STATE", "CONTRACT_TITLE"])
# Verify PAYER_STATE (list[str] in mapping) is normalized
assert result["PAYER_STATE"] == ["IL"]
assert isinstance(result["PAYER_STATE"], list)
# Verify CONTRACT_TITLE (str in mapping) is normalized from list to str
assert result["CONTRACT_TITLE"] == "My Contract"
assert isinstance(result["CONTRACT_TITLE"], str)
# Verify LOB (list[str] in mapping) is NOT normalized (not in field_names)
assert result["LOB"] == "Medicare" # Preserved as string, not normalized
assert isinstance(result["LOB"], str)
# Verify PROV_INFO_JSON (list[dict] in mapping) is NOT normalized (not in field_names)
assert isinstance(result["PROV_INFO_JSON"], list)
assert len(result["PROV_INFO_JSON"]) == 1
# Test without field_names - all fields in mapping should be normalized
result_all = parse_json_dict(output)
# All fields in mapping should be normalized
assert result_all["PAYER_STATE"] == ["IL"]
assert isinstance(result_all["PAYER_STATE"], list)
assert result_all["CONTRACT_TITLE"] == "My Contract"
assert isinstance(result_all["CONTRACT_TITLE"], str)
assert result_all["LOB"] == ["Medicare"] # Normalized to list
assert isinstance(result_all["LOB"], list)
assert isinstance(result_all["PROV_INFO_JSON"], list)
assert len(result_all["PROV_INFO_JSON"]) == 1
class TestParseJsonList:
@@ -145,51 +241,38 @@ class TestParseJsonList:
def test_list_with_explanation_after(self):
"""Test parsing JSON list with explanatory text after it."""
output = '["item1", "item2"] as you can see above'
output = '["key", "value"] This is some explanation text.'
result = parse_json_list(output)
assert result == ["item1", "item2"]
assert result == ["key", "value"]
def test_list_with_explanation_both_sides(self):
"""Test parsing JSON list with text on both sides."""
output = 'Results: ["YES"] based on analysis'
output = 'Analysis: The answer is ["result", "success"] as shown above.'
result = parse_json_list(output)
assert result == ["YES"]
def test_list_of_numbers(self):
"""Test parsing list of numbers."""
output = "[1, 2, 3, 4, 5]"
result = parse_json_list(output)
assert result == [1, 2, 3, 4, 5]
def test_list_of_mixed_types(self):
"""Test parsing list with mixed types."""
output = '["string", 42, true, null]'
result = parse_json_list(output)
assert result == ["string", 42, True, None]
assert result == ["result", "success"]
def test_nested_list(self):
"""Test parsing nested JSON list."""
output = '[["a", "b"], ["c", "d"]]'
output = "[[1, 2], [3, 4], [5, 6]]"
result = parse_json_list(output)
assert result == [["a", "b"], ["c", "d"]]
assert result == [[1, 2], [3, 4], [5, 6]]
def test_list_of_dicts(self):
"""Test parsing list containing dictionaries."""
output = '[{"field": "value1"}, {"field": "value2"}]'
def test_list_with_special_characters(self):
"""Test parsing list with special characters in strings."""
output = '["value with \\"quotes\\"", "other value"]'
result = parse_json_list(output)
assert result == [{"field": "value1"}, {"field": "value2"}]
assert result == ['value with "quotes"', "other value"]
def test_list_with_newlines(self):
"""Test parsing list with newlines in the JSON."""
output = """
[
"item1",
"item2",
"item3"
"value1",
"value2"
]
"""
result = parse_json_list(output)
assert result == ["item1", "item2", "item3"]
assert result == ["value1", "value2"]
def test_multiple_lists_returns_last(self):
"""Test that when multiple lists present, the last one is returned."""
@@ -197,149 +280,237 @@ class TestParseJsonList:
result = parse_json_list(output)
assert result == ["second"]
def test_list_with_null_values(self):
"""Test parsing list with null values."""
output = '["value", null, "other"]'
result = parse_json_list(output)
assert result == ["value", None, "other"]
def test_list_with_boolean_values(self):
"""Test parsing list with boolean values."""
output = "[true, false, true]"
result = parse_json_list(output)
assert result == [True, False, True]
def test_list_with_numeric_values(self):
"""Test parsing list with various numeric types."""
output = "[42, 3.14, -10]"
result = parse_json_list(output)
assert result == [42, 3.14, -10]
def test_empty_list(self):
"""Test parsing an empty list."""
output = "[]"
result = parse_json_list(output)
assert result == []
def test_single_item_list(self):
"""Test parsing list with single item."""
output = '["N/A"]'
def test_list_with_unicode(self):
"""Test parsing list with unicode characters."""
output = '["café", "😀"]'
result = parse_json_list(output)
assert result == ["N/A"]
assert result == ["café", "😀"]
def test_no_list_raises_error(self):
"""Test that ValueError is raised when no list is found."""
output = "This is just text without any JSON"
with pytest.raises(ValueError, match="No valid JSON list found"):
parse_json_list(output)
def test_list_with_dicts(self):
"""Test parsing list containing dictionaries."""
output = '[{"key1": "value1"}, {"key2": "value2"}]'
result = parse_json_list(output)
assert result == [{"key1": "value1"}, {"key2": "value2"}]
def test_only_dict_raises_error(self):
"""Test that ValueError is raised when only a dict is found."""
output = '{"field": "value"}'
with pytest.raises(ValueError, match="No valid JSON list found"):
parse_json_list(output)
def test_list_with_field_name_normalization(self):
"""Test list normalization with field_name."""
# PAYER_STATE is list[str] in mapping
output = '["IL", "CA"]'
result = parse_json_list(output, field_name="PAYER_STATE")
assert result == ["IL", "CA"]
assert all(isinstance(item, str) for item in result)
def test_malformed_json_raises_error(self):
"""Test that ValueError is raised for malformed JSON."""
output = '["item1", "item2"' # Missing closing bracket
with pytest.raises(ValueError, match="No valid JSON list found"):
parse_json_list(output)
def test_non_string_input_raises_error(self):
"""Test that TypeError is raised for non-string input."""
with pytest.raises(TypeError, match="Expected string input"):
parse_json_list([1, 2, 3])
class TestParseJsonDictOrList:
"""Tests for parse_json_dict_or_list function."""
def test_returns_dict(self):
"""Test that it returns a dict when dict is present."""
output = '{"field": "value"}'
result = parse_json_dict_or_list(output)
assert result == {"field": "value"}
assert isinstance(result, dict)
def test_returns_list(self):
"""Test that it returns a list when list is present."""
output = '["value1", "value2"]'
result = parse_json_dict_or_list(output)
assert result == ["value1", "value2"]
def test_list_with_expected_format(self):
"""Test list normalization with expected_format."""
output = '[{"NAME": "Provider"}]'
result = parse_json_list(
output, field_name="PROV_INFO_JSON", expected_format="list[dict[str,str]]"
)
assert isinstance(result, list)
def test_returns_last_when_both_present(self):
"""Test that it returns the last JSON object when both types present."""
output = '{"first": 1} and then ["second"]'
result = parse_json_dict_or_list(output)
assert result == ["second"]
assert isinstance(result, list)
def test_returns_last_dict_when_multiple_dicts(self):
"""Test that it returns the last dict when multiple dicts present."""
output = '{"first": 1} and {"second": 2}'
result = parse_json_dict_or_list(output)
assert result == {"second": 2}
def test_no_json_raises_error(self):
"""Test that ValueError is raised when no JSON is found."""
output = "No JSON here"
with pytest.raises(ValueError, match="No valid JSON object found"):
parse_json_dict_or_list(output)
def test_non_string_input_raises_error(self):
"""Test that TypeError is raised for non-string input."""
with pytest.raises(TypeError, match="Expected string input"):
parse_json_dict_or_list(None)
assert len(result) == 1
assert isinstance(result[0], dict)
class TestEdgeCases:
"""Tests for edge cases and special scenarios."""
"""Tests for edge cases and error handling."""
def test_json_with_curly_braces_in_strings(self):
"""Test parsing JSON containing braces in string values."""
output = '{"template": "Use {variable} here"}'
def test_invalid_json_dict(self):
"""Test that invalid JSON raises ValueError."""
output = '{"invalid": json}'
with pytest.raises(ValueError):
parse_json_dict(output)
def test_invalid_json_list(self):
"""Test that invalid JSON raises ValueError."""
output = "[invalid json]"
with pytest.raises(ValueError):
parse_json_list(output)
def test_non_string_input_dict(self):
"""Test that non-string input raises TypeError."""
with pytest.raises(TypeError):
parse_json_dict(123)
def test_non_string_input_list(self):
"""Test that non-string input raises TypeError."""
with pytest.raises(TypeError):
parse_json_list(123)
def test_empty_string_dict(self):
"""Test that empty string raises ValueError."""
with pytest.raises(ValueError):
parse_json_dict("")
def test_empty_string_list(self):
"""Test that empty string raises ValueError."""
with pytest.raises(ValueError):
parse_json_list("")
def test_dict_with_list_values(self):
"""Test dict with list values that need normalization."""
output = '{"PAYER_STATE": "IL"}'
result = parse_json_dict(output, field_names=["PAYER_STATE"])
# PAYER_STATE is list[str] in mapping, so string should become list
assert result["PAYER_STATE"] == ["IL"]
assert isinstance(result["PAYER_STATE"], list)
class TestFieldAwareNormalization:
"""Tests for field-aware normalization functionality."""
def test_normalize_str_field_from_string(self):
"""Test that str fields remain strings."""
output = '{"CONTRACT_TITLE": "My Contract"}'
result = parse_json_dict(output, field_names=["CONTRACT_TITLE"])
assert result["CONTRACT_TITLE"] == "My Contract"
assert isinstance(result["CONTRACT_TITLE"], str)
def test_normalize_str_field_from_list(self):
"""Test that str fields convert single-element lists to strings."""
output = '{"CONTRACT_TITLE": ["My Contract"]}'
result = parse_json_dict(output, field_names=["CONTRACT_TITLE"])
assert result["CONTRACT_TITLE"] == "My Contract"
assert isinstance(result["CONTRACT_TITLE"], str)
def test_normalize_list_str_field_from_string(self):
"""Test that list[str] fields convert strings to lists."""
output = '{"PAYER_STATE": "IL"}'
result = parse_json_dict(output, field_names=["PAYER_STATE"])
assert result["PAYER_STATE"] == ["IL"]
assert isinstance(result["PAYER_STATE"], list)
assert all(isinstance(item, str) for item in result["PAYER_STATE"])
def test_normalize_list_str_field_from_list(self):
"""Test that list[str] fields keep lists as lists."""
output = '{"PAYER_STATE": ["IL", "CA"]}'
result = parse_json_dict(output, field_names=["PAYER_STATE"])
assert result["PAYER_STATE"] == ["IL", "CA"]
assert isinstance(result["PAYER_STATE"], list)
def test_normalize_list_dict_field(self):
"""Test normalization of list[dict[str,str]] fields."""
output = '{"PROV_INFO_JSON": [{"NAME": "Provider", "TIN": "123"}]}'
result = parse_json_dict(output, field_names=["PROV_INFO_JSON"])
assert isinstance(result["PROV_INFO_JSON"], list)
assert len(result["PROV_INFO_JSON"]) == 1
assert isinstance(result["PROV_INFO_JSON"][0], dict)
assert result["PROV_INFO_JSON"][0] == {"NAME": "Provider", "TIN": "123"}
def test_normalize_string_representation_of_list(self):
"""Test normalization of string representation of list."""
# Sometimes LLM returns string representation of list
output = "{\"PAYER_STATE\": \"['IL', 'CA']\"}"
result = parse_json_dict(output, field_names=["PAYER_STATE"])
assert result["PAYER_STATE"] == ["IL", "CA"]
assert isinstance(result["PAYER_STATE"], list)
def test_normalize_string_representation_of_dict(self):
"""Test normalization of string representation of dict."""
output = '{"PROV_INFO_JSON": "[{\\"NAME\\": \\"Provider\\"}]"}'
result = parse_json_dict(output, field_names=["PROV_INFO_JSON"])
assert isinstance(result["PROV_INFO_JSON"], list)
assert len(result["PROV_INFO_JSON"]) == 1
assert isinstance(result["PROV_INFO_JSON"][0], dict)
def test_list_normalization_with_field_name(self):
"""Test list normalization when field_name is provided."""
output = '["IL", "CA"]'
result = parse_json_list(output, field_name="PAYER_STATE")
assert result == ["IL", "CA"]
assert all(isinstance(item, str) for item in result)
def test_list_normalization_without_field_name(self):
"""Test list normalization when field_name is not provided."""
output = '[{"NAME": "Provider"}]'
result = parse_json_list(output) # No field_name
# Should normalize dicts in the list
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], dict)
def test_normalize_pipe_delimited_string(self):
"""Test normalization of pipe-delimited string to list."""
output = '{"PAYER_STATE": "IL|CA"}'
result = parse_json_dict(output, field_names=["PAYER_STATE"])
assert result["PAYER_STATE"] == ["IL", "CA"]
assert isinstance(result["PAYER_STATE"], list)
def test_normalization_handles_empty_values(self):
"""Test normalization of empty values."""
output = '{"PAYER_STATE": ""}'
result = parse_json_dict(output, field_names=["PAYER_STATE"])
assert result["PAYER_STATE"] == []
def test_normalization_handles_none_values(self):
"""Test normalization of None values."""
output = '{"PAYER_STATE": null}'
result = parse_json_dict(output, field_names=["PAYER_STATE"])
assert result["PAYER_STATE"] == []
def test_normalization_handles_missing_fields(self):
"""Test that missing fields don't cause errors."""
output = '{"OTHER_FIELD": "value"}'
result = parse_json_dict(output, field_names=["PAYER_STATE"])
# PAYER_STATE not in output, should not cause error
assert "OTHER_FIELD" in result
def test_dict_normalization_with_vs_without_field_names(self):
"""Test difference between providing field_names vs not."""
output = '{"PAYER_STATE": "IL", "UNKNOWN_FIELD": ["a", "b"]}'
# With field_names - only specified fields normalized
result_with = parse_json_dict(output, field_names=["PAYER_STATE"])
assert result_with["PAYER_STATE"] == ["IL"] # Normalized (in mapping)
# UNKNOWN_FIELD not in field_names and not in mapping, so preserved as-is
assert result_with["UNKNOWN_FIELD"] == ["a", "b"] # Preserved
# Without field_names - only fields in mapping are normalized
result_without = parse_json_dict(output)
assert result_without["PAYER_STATE"] == ["IL"] # Normalized (in mapping)
assert result_without["UNKNOWN_FIELD"] == [
"a",
"b",
] # Preserved (not in mapping)
def test_list_normalization_empty_list(self):
"""Test normalization of empty list."""
output = "[]"
result = parse_json_list(output, field_name="PAYER_STATE")
assert result == []
def test_list_normalization_single_item(self):
"""Test normalization of single-item list."""
output = '["IL"]'
result = parse_json_list(output, field_name="PAYER_STATE")
assert result == ["IL"]
def test_dict_normalization_preserves_order(self):
"""Test that field order is preserved during normalization."""
output = '{"A": "1", "B": "2", "C": "3"}'
result = parse_json_dict(output)
assert result == {"template": "Use {variable} here"}
def test_json_with_brackets_in_strings(self):
"""Test parsing JSON containing brackets in string values."""
output = '["[optional]", "[required]"]'
result = parse_json_list(output)
assert result == ["[optional]", "[required]"]
def test_json_with_escaped_quotes(self):
"""Test parsing JSON with escaped quotes in strings."""
output = '{"quote": "He said \\"hello\\""}'
result = parse_json_dict(output)
assert result == {"quote": 'He said "hello"'}
def test_json_with_backslashes(self):
"""Test parsing JSON with backslashes."""
output = '{"path": "C:\\\\Users\\\\file.txt"}'
result = parse_json_dict(output)
assert result == {"path": "C:\\Users\\file.txt"}
def test_very_long_string_values(self):
"""Test parsing JSON with very long string values."""
long_text = "x" * 10000
output = f'{{"text": "{long_text}"}}'
result = parse_json_dict(output)
assert result == {"text": long_text}
def test_deeply_nested_structure(self):
"""Test parsing deeply nested JSON structure."""
output = '{"a": {"b": {"c": {"d": "value"}}}}'
result = parse_json_dict(output)
assert result == {"a": {"b": {"c": {"d": "value"}}}}
def test_json_in_markdown_code_block(self):
"""Test parsing JSON within markdown code blocks."""
output = """
```json
{"field": "value"}
```
"""
result = parse_json_dict(output)
assert result == {"field": "value"}
def test_json_with_reasoning_text(self):
"""Test realistic LLM output with reasoning and JSON."""
output = """
Based on the analysis, I found the following information:
1. The exhibit clearly states the LOB
2. Multiple programs are mentioned
Here is my answer:
{"LOB": ["Medicare", "Duals"], "PROGRAM": ["MA", "MASNP"]}
"""
result = parse_json_dict(output)
assert result == {"LOB": ["Medicare", "Duals"], "PROGRAM": ["MA", "MASNP"]}
def test_json_with_explanation_in_chinese(self):
"""Test parsing with non-English explanatory text."""
output = '这是答案: {"field": "value"}'
result = parse_json_dict(output)
assert result == {"field": "value"}
keys = list(result.keys())
assert keys == ["A", "B", "C"]
src/tests/test_prompt_calls.py
+1019
View File
File diff suppressed because it is too large Load Diff
src/utils/formatting_utils.py
+265
View File
@@ -0,0 +1,265 @@
"""
Format normalization utilities for LLM output.
This module provides functions to normalize field values to their expected
format types based on the field format mapping configuration.
"""
import ast
import json
import logging
from typing import Any
from src.constants.investment_columns import FIELD_FORMAT_MAPPING
logger = logging.getLogger(__name__)
def normalize_field_value(field_name: str, value: Any, format_type: str) -> Any:
"""
Normalize a field value to the expected format type.
Args:
field_name: Name of the field being normalized (for logging)
value: The value to normalize (can be str, list, dict, etc.)
format_type: Expected format type ('str', 'list[str]', 'list[dict[str,str]]', etc.)
Returns:
Normalized value in the expected format
Examples:
normalize_field_value("PAYER_STATE", "IL", "list[str]") -> ["IL"]
normalize_field_value("PAYER_STATE", ["IL", "CA"], "list[str]") -> ["IL", "CA"]
normalize_field_value("CONTRACT_TITLE", ["Title"], "str") -> "Title"
normalize_field_value("PROV_INFO_JSON", {...}, "list[dict[str,str]]") -> [{...}]
"""
if value is None:
return _get_default_for_format(format_type)
# Handle string representations of lists/dicts first
if isinstance(value, str) and value.strip():
# Try to parse as JSON first
if value.strip().startswith(("[", "{")):
try:
parsed = json.loads(value)
return normalize_field_value(field_name, parsed, format_type)
except (json.JSONDecodeError, ValueError):
pass
# Try to parse as Python literal
try:
parsed = ast.literal_eval(value)
if parsed != value: # Only use if it actually parsed something different
return normalize_field_value(field_name, parsed, format_type)
except (ValueError, SyntaxError):
pass
# Normalize based on format type
if format_type == "str":
return _normalize_to_str(value)
elif format_type == "list[str]":
return _normalize_to_list_str(value)
elif format_type == "list[dict[str,str]]":
return _normalize_to_list_dict(value)
elif format_type == "dict[str,str]":
return _normalize_to_dict(value)
else:
logger.warning(
f"Unknown format type '{format_type}' for field '{field_name}'. "
f"Returning value as-is."
)
return value
def _normalize_to_str(value: Any) -> str:
"""Normalize value to a string."""
if isinstance(value, str):
return value
elif isinstance(value, list):
if len(value) == 0:
return ""
elif len(value) == 1:
return str(value[0])
else:
# Join multiple values with a delimiter (pipe for backward compatibility)
return "|".join(str(v) for v in value)
elif isinstance(value, dict):
# Convert dict to string representation
return json.dumps(value)
else:
return str(value) if value is not None else ""
def _normalize_to_list_str(value: Any) -> list[str]:
"""Normalize value to a list of strings."""
if isinstance(value, list):
# Ensure all elements are strings
return [str(v) if v is not None else "" for v in value]
elif isinstance(value, str):
if not value or value.strip() == "":
return []
# Try to parse as JSON list first
if value.strip().startswith("["):
try:
parsed = json.loads(value)
if isinstance(parsed, list):
return [str(v) if v is not None else "" for v in parsed]
except (json.JSONDecodeError, ValueError):
pass
# Try to parse as Python literal
try:
parsed = ast.literal_eval(value)
if isinstance(parsed, list):
return [str(v) if v is not None else "" for v in parsed]
except (ValueError, SyntaxError):
pass
# Split on pipe delimiter (backward compatibility)
if "|" in value:
return [v.strip() for v in value.split("|") if v.strip()]
# Single string value - wrap in list
return [value]
elif isinstance(value, dict):
# Convert dict values to list
return [str(v) if v is not None else "" for v in value.values()]
else:
return [str(value)] if value is not None else []
def _normalize_to_list_dict(value: Any) -> list[dict[str, str]]:
"""Normalize value to a list of dictionaries with string keys and values."""
if isinstance(value, list):
result = []
for item in value:
if isinstance(item, dict):
# Ensure all values are strings
normalized_dict = {
str(k): str(v) if v is not None else "" for k, v in item.items()
}
result.append(normalized_dict)
elif isinstance(item, str):
# Try to parse string as dict
try:
parsed = json.loads(item)
if isinstance(parsed, dict):
normalized_dict = {
str(k): str(v) if v is not None else ""
for k, v in parsed.items()
}
result.append(normalized_dict)
except (json.JSONDecodeError, ValueError):
logger.warning(
f"Could not parse string as dict in list: {item[:50]}..."
)
return result
elif isinstance(value, dict):
# Single dict - wrap in list
normalized_dict = {
str(k): str(v) if v is not None else "" for k, v in value.items()
}
return [normalized_dict]
elif isinstance(value, str):
if not value or value.strip() == "":
return []
# Try to parse as JSON
try:
parsed = json.loads(value)
return _normalize_to_list_dict(parsed)
except (json.JSONDecodeError, ValueError):
try:
parsed = ast.literal_eval(value)
return _normalize_to_list_dict(parsed)
except (ValueError, SyntaxError):
logger.warning(f"Could not parse string as list[dict]: {value[:50]}...")
return []
else:
return []
def _normalize_to_dict(value: Any) -> dict[str, str]:
"""Normalize value to a dictionary with string keys and values."""
if isinstance(value, dict):
return {str(k): str(v) if v is not None else "" for k, v in value.items()}
elif isinstance(value, str):
if not value or value.strip() == "":
return {}
try:
parsed = json.loads(value)
if isinstance(parsed, dict):
return {
str(k): str(v) if v is not None else "" for k, v in parsed.items()
}
except (json.JSONDecodeError, ValueError):
try:
parsed = ast.literal_eval(value)
if isinstance(parsed, dict):
return {
str(k): str(v) if v is not None else ""
for k, v in parsed.items()
}
except (ValueError, SyntaxError):
pass
return {}
elif isinstance(value, list):
# If list has one dict, return it
if len(value) == 1 and isinstance(value[0], dict):
return {
str(k): str(v) if v is not None else "" for k, v in value[0].items()
}
return {}
else:
return {}
def _get_default_for_format(format_type: str) -> Any:
"""Get default value for a given format type."""
if format_type == "str":
return ""
elif format_type == "list[str]":
return []
elif format_type == "list[dict[str,str]]":
return []
elif format_type == "dict[str,str]":
return {}
else:
return None
def normalize_dict_fields(
parsed_dict: dict[str, Any], field_names: list[str] | None = None
) -> dict[str, Any]:
"""
Normalize all fields in a dictionary based on field format mapping.
Args:
parsed_dict: Dictionary with parsed LLM output
field_names: Optional list of field names to normalize. If None, normalizes
all fields that exist in the format mapping. Fields not in the mapping
are left unchanged.
Returns:
Dictionary with normalized field values
"""
normalized = {}
fields_to_normalize = field_names if field_names else list(parsed_dict.keys())
for field_name in fields_to_normalize:
if field_name not in parsed_dict:
continue
# Only normalize if field is in the format mapping
if field_name in FIELD_FORMAT_MAPPING:
format_type = FIELD_FORMAT_MAPPING[field_name]
value = parsed_dict[field_name]
normalized[field_name] = normalize_field_value(
field_name, value, format_type
)
else:
# Field not in mapping, preserve original value
normalized[field_name] = parsed_dict[field_name]
# Preserve any fields not processed
for key, value in parsed_dict.items():
if key not in normalized:
normalized[key] = value
return normalized
src/utils/json_utils.py
+85 -84
View File
@@ -3,6 +3,9 @@ JSON parsing utilities for LLM responses.
This module provides robust parsers for extracting JSON dictionaries and lists
from raw LLM output, replacing the deprecated pipe-delimited format.
The parsers can optionally normalize field values based on the field format
mapping configuration to ensure consistent output formats.
"""
import json
@@ -10,11 +13,15 @@ import logging
import re
from typing import Any
from src.utils.formatting_utils import normalize_dict_fields
logger = logging.getLogger(__name__)
def parse_json_dict(raw_llm_output: str) -> dict[str, Any]:
def parse_json_dict(
raw_llm_output: str, field_names: list[str] | None = None
) -> dict[str, Any]:
"""
Extract and parse a JSON dictionary from raw LLM output.
@@ -22,12 +29,18 @@ def parse_json_dict(raw_llm_output: str) -> dict[str, Any]:
raw output string and returns it as a Python dict. It handles common cases
where the LLM includes explanatory text before or after the JSON.
Optionally normalizes field values based on the field format mapping
configuration to ensure consistent output formats.
Args:
raw_llm_output: Raw string output from the LLM, potentially containing
explanatory text along with JSON dictionary.
field_names: Optional list of field names to normalize. If provided,
field values will be normalized according to the field format mapping.
Returns:
dict: The parsed JSON dictionary.
dict: The parsed JSON dictionary, with normalized field values if
field_names is provided.
Raises:
ValueError: If no valid JSON dictionary is found in the output.
@@ -38,9 +51,9 @@ def parse_json_dict(raw_llm_output: str) -> dict[str, Any]:
>>> parse_json_dict(output)
{'field': 'value', 'count': 42}
>>> output = '{"key": "value"} This is some explanation text.'
>>> parse_json_dict(output)
{'key': 'value'}
>>> output = '{"PAYER_STATE": "IL"}'
>>> parse_json_dict(output, field_names=["PAYER_STATE"])
{'PAYER_STATE': ['IL']} # Normalized to list[str]
"""
if not isinstance(raw_llm_output, str):
raise TypeError(f"Expected string input, got {type(raw_llm_output).__name__}")
@@ -91,7 +104,18 @@ def parse_json_dict(raw_llm_output: str) -> dict[str, Any]:
if found_dicts:
# Return the last valid dictionary found
return found_dicts[-1]
parsed_dict = found_dicts[-1]
# Always normalize field values based on format mapping
# If field_names provided, normalize only those fields
# Otherwise, normalize all fields found in the parsed dict
if field_names:
parsed_dict = normalize_dict_fields(parsed_dict, field_names)
else:
# Normalize all fields in the parsed dict
parsed_dict = normalize_dict_fields(parsed_dict, field_names=None)
return parsed_dict
raise ValueError(
f"No valid JSON dictionary found in LLM output. "
@@ -99,7 +123,11 @@ def parse_json_dict(raw_llm_output: str) -> dict[str, Any]:
)
def parse_json_list(raw_llm_output: str) -> list[Any]:
def parse_json_list(
raw_llm_output: str,
field_name: str | None = None,
expected_format: str | None = None,
) -> list[Any]:
"""
Extract and parse a JSON list (array) from raw LLM output.
@@ -107,12 +135,19 @@ def parse_json_list(raw_llm_output: str) -> list[Any]:
string and returns it as a Python list. It handles common cases where
the LLM includes explanatory text before or after the JSON.
Optionally normalizes the list contents based on the field format mapping
if the list contains dictionaries that need field normalization.
Args:
raw_llm_output: Raw string output from the LLM, potentially containing
explanatory text along with JSON list.
field_name: Optional field name for normalization. If the list contains
dictionaries, their fields will be normalized based on the format mapping.
expected_format: Optional expected format type (e.g., 'list[dict[str,str]]').
If not provided, will be looked up from field format mapping using field_name.
Returns:
list: The parsed JSON list.
list: The parsed JSON list, with normalized contents if field_name is provided.
Raises:
ValueError: If no valid JSON list is found in the output.
@@ -123,9 +158,9 @@ def parse_json_list(raw_llm_output: str) -> list[Any]:
>>> parse_json_list(output)
['value1', 'value2', 'value3']
>>> output = '["item1", "item2"] as you can see above'
>>> parse_json_list(output)
['item1', 'item2']
>>> output = '[{"TIN": "123", "NPI": "456"}]'
>>> parse_json_list(output, field_name="PROV_INFO_JSON")
[{'TIN': '123', 'NPI': '456'}] # Normalized if needed
"""
if not isinstance(raw_llm_output, str):
raise TypeError(f"Expected string input, got {type(raw_llm_output).__name__}")
@@ -176,81 +211,47 @@ def parse_json_list(raw_llm_output: str) -> list[Any]:
if found_lists:
# Return the last valid list found
return found_lists[-1]
parsed_list = found_lists[-1]
# Always normalize list contents based on format mapping
# If field_name provided, use it to determine format
# Otherwise, try to infer format from list contents
from src.constants.investment_columns import FIELD_FORMAT_MAPPING
from src.utils.formatting_utils import normalize_field_value
if field_name:
# Check if field is in mapping
if field_name in FIELD_FORMAT_MAPPING:
format_type = expected_format or FIELD_FORMAT_MAPPING[field_name]
normalized_list = normalize_field_value(
field_name, parsed_list, format_type
)
return normalized_list
else:
# Field not in mapping, but if expected_format provided, use it
if expected_format:
normalized_list = normalize_field_value(
field_name, parsed_list, expected_format
)
return normalized_list
# Otherwise return as-is
return parsed_list
else:
# If no field_name, check if list contains dicts and normalize them
if parsed_list and isinstance(parsed_list[0], dict):
# Normalize each dict in the list
normalized_list = []
for item in parsed_list:
if isinstance(item, dict):
normalized_item = normalize_dict_fields(item, field_names=None)
normalized_list.append(normalized_item)
else:
normalized_list.append(item)
return normalized_list
return parsed_list
raise ValueError(
f"No valid JSON list found in LLM output. "
f"Output preview: {raw_llm_output[:200]}..."
)
def parse_json_dict_or_list(raw_llm_output: str) -> dict[str, Any] | list[Any]:
"""
Extract and parse either a JSON dictionary or list from raw LLM output.
This is a convenience function that attempts to parse the output as either
a dictionary or a list, returning whichever is found last in the output.
Useful when the expected output format may vary.
Args:
raw_llm_output: Raw string output from the LLM.
Returns:
dict | list: The parsed JSON object (dictionary or list).
Raises:
ValueError: If no valid JSON dictionary or list is found in the output.
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__}")
found_objects = []
i = 0
while i < len(raw_llm_output):
if raw_llm_output[i] in "{[":
start = i
stack = [raw_llm_output[i]]
j = i + 1
while j < len(raw_llm_output):
# Skip over string literals
if raw_llm_output[j] == '"':
j += 1
while j < len(raw_llm_output):
if raw_llm_output[j] == "\\" and j + 1 < len(raw_llm_output):
j += 2
continue
if raw_llm_output[j] == '"':
break
j += 1
elif raw_llm_output[j] in "{[":
stack.append(raw_llm_output[j])
elif raw_llm_output[j] in "}]":
if not stack:
break
open_bracket = stack.pop()
if (open_bracket == "{" and raw_llm_output[j] != "}") or (
open_bracket == "[" and raw_llm_output[j] != "]"
):
break
if not stack:
candidate = raw_llm_output[start : j + 1]
try:
obj = json.loads(candidate)
found_objects.append(obj)
except json.JSONDecodeError:
pass
i = j
break
j += 1
i += 1
if found_objects:
return found_objects[-1]
raise ValueError(
f"No valid JSON object found in LLM output. "
f"Output preview: {raw_llm_output[:200]}..."
)
src/utils/string_utils.py
-98
View File
@@ -806,101 +806,3 @@ def normalize_to_json_list(val):
result = json.dumps(expanded)
return result
return val
def normalize_one_to_one_field_value(field_name: str, value) -> str | list:
"""
Normalize a 1:1 field value to ensure single-value fields are strings, not lists.
This function handles the common case where LLM returns a single-value field as
a list with one element (e.g., ['The Value']). It extracts the string from such
lists while preserving legitimate multi-value lists.
Args:
field_name (str): Name of the field being normalized (for logging/debugging)
value: The field value to normalize. Can be:
- str: Returned as-is
- list with 1 element: Extract the string
- list with >1 elements: Returned unchanged (may be legitimate multi-value)
- None/empty: Returned as-is
Returns:
str | list: Normalized value:
- If input is a list with exactly 1 element: returns that element as string
- If input is already a string: returns as-is
- If input is a list with >1 elements: returns list unchanged
- If input is None or empty: returns as-is
Examples:
>>> normalize_one_to_one_field_value("CONTRACT_TITLE", ["Provider Agreement"])
'Provider Agreement'
>>> normalize_one_to_one_field_value("PAYER_NAME", "Aetna")
'Aetna'
>>> normalize_one_to_one_field_value("PROVIDER_NAME", ["Provider A", "Provider B"])
['Provider A', 'Provider B'] # Multi-value preserved
>>> normalize_one_to_one_field_value("LOB", ["Medicare", "Medicaid"])
['Medicare', 'Medicaid'] # Multi-value preserved
"""
# Handle None and empty values
if value is None or (isinstance(value, str) and is_empty(value)):
return value
# If it's already a string, return as-is
if isinstance(value, str):
return value
# If it's a list
if isinstance(value, list):
# List with exactly 1 element: extract the string
if len(value) == 1:
return str(value[0]) if value[0] is not None else "N/A"
# List with >1 elements: preserve (may be legitimate multi-value)
# This handles cases like PROVIDER_NAME with multiple providers,
# or dynamic fields (LOB, PROGRAM, PRODUCT, NETWORK) when passed to 1:1
elif len(value) > 1:
return value
# Empty list: return "N/A"
else:
return "N/A"
# For any other type, convert to string
return str(value) if value is not None else "N/A"
def normalize_one_to_one_answers_dict(
answers_dict: dict[str, str | list],
) -> dict[str, str | list]:
"""
Normalize all values in a 1:1 answers dictionary.
Applies normalization to all fields in the dictionary, converting single-element
lists to strings while preserving legitimate multi-value lists.
Args:
answers_dict (dict): Dictionary of field names to values (may be strings or lists)
Returns:
dict: Dictionary with normalized values (single-element lists converted to strings)
Example:
>>> answers = {
... "CONTRACT_TITLE": ["Provider Agreement"],
... "PAYER_NAME": "Aetna",
... "PROVIDER_NAME": ["Provider A", "Provider B"]
... }
>>> normalize_one_to_one_answers_dict(answers)
{
"CONTRACT_TITLE": "Provider Agreement",
"PAYER_NAME": "Aetna",
"PROVIDER_NAME": ["Provider A", "Provider B"]
}
"""
DO_NOT_NORMALIZE = ["PROV_INFO_JSON", "PAYER_STATE", "PROVIDER_STATE"]
normalized = {}
for field_name, value in answers_dict.items():
if field_name not in DO_NOT_NORMALIZE:
normalized[field_name] = normalize_one_to_one_field_value(field_name, value)
else:
normalized[field_name] = value
return normalized