diff --git a/src/pipelines/runner.py b/src/pipelines/runner.py index b13de73..d8f6c42 100644 --- a/src/pipelines/runner.py +++ b/src/pipelines/runner.py @@ -212,23 +212,23 @@ def main(client: str = "saas", testing=False, test_params={}): error_df = pd.DataFrame([{"error": str(e)}]) error_results.append(error_df) - # Combine results - if len(successful_results_cc) > 0: - FINAL_RESULT_DF_CC = pd.concat(successful_results_cc, ignore_index=True) - if successful_results_dashboard: - FINAL_RESULT_DF_DASHBOARD = pd.concat( - successful_results_dashboard, ignore_index=True - ) - else: - FINAL_RESULT_DF_DASHBOARD = pd.DataFrame() + # Combine results + if len(successful_results_cc) > 0: + FINAL_RESULT_DF_CC = pd.concat(successful_results_cc, ignore_index=True) + if successful_results_dashboard: + FINAL_RESULT_DF_DASHBOARD = pd.concat( + successful_results_dashboard, ignore_index=True + ) else: - FINAL_RESULT_DF_CC = pd.DataFrame() FINAL_RESULT_DF_DASHBOARD = pd.DataFrame() + else: + FINAL_RESULT_DF_CC = pd.DataFrame() + FINAL_RESULT_DF_DASHBOARD = pd.DataFrame() - if len(error_results) > 0: - ERROR_RESULT_DF = pd.concat(error_results, ignore_index=True) - else: - ERROR_RESULT_DF = pd.DataFrame() + if len(error_results) > 0: + ERROR_RESULT_DF = pd.concat(error_results, ignore_index=True) + else: + ERROR_RESULT_DF = pd.DataFrame() # ========== COMMON: QC/QA Validation ========== # Run QC/QA validation by default (preserves automatic behavior for single files and batches) diff --git a/src/pipelines/saas/prompts/prompt_calls.py b/src/pipelines/saas/prompts/prompt_calls.py index 3c0f53d..0ff35c7 100644 --- a/src/pipelines/saas/prompts/prompt_calls.py +++ b/src/pipelines/saas/prompts/prompt_calls.py @@ -87,10 +87,12 @@ def prompt_exhibit_level_breakout( def prompt_dynamic_primary( exhibit_text: str, field: Field, constants: Constants, filename: str, TEMPLATE ): + """Extract dynamic primary field from exhibit text.""" prompt, _parser = TEMPLATE( exhibit_text, field.field_name, field.get_prompt(constants) ) logging.debug(f"Dynamic primary prompt for {filename}; {field}: {prompt}") + llm_answer_raw = llm_utils.invoke_claude( prompt, "sonnet_latest", @@ -99,9 +101,8 @@ def prompt_dynamic_primary( instruction=prompt_templates.DYNAMIC_PRIMARY_INSTRUCTION(), ) logging.debug(f"Claude answer for {filename}; {field}: {llm_answer_raw}") - llm_answer_final = _parser(llm_answer_raw) - # Normalization is already done in the JSON parser + llm_answer_final = _parser(llm_answer_raw) return llm_answer_final @@ -411,13 +412,15 @@ def prompt_full_context( # Extract field names for field-aware normalization field_names = full_context_fields.list_fields() + context_text = contract_text[ + 0 : min( + config.MAX_CONTEXT_LENGTH - len(prompt_questions), + len(contract_text) - 1, + ) + ] + full_context_prompt, _parser = prompt_templates.ONE_TO_ONE_MULTI_FIELD_TEMPLATE( - context=contract_text[ - 0 : min( - config.MAX_CONTEXT_LENGTH - len(prompt_questions), - len(contract_text) - 1, - ) - ], + context=context_text, questions=prompt_questions, field_names=field_names, ) @@ -433,6 +436,7 @@ def prompt_full_context( ) full_context_answers_dict = _parser(claude_answer_raw) + # Field-aware normalization is already done in the JSON parser # No additional normalization needed for fields in FIELD_FORMAT_MAPPING except Exception as e: diff --git a/src/pipelines/shared/extraction/dynamic_funcs.py b/src/pipelines/shared/extraction/dynamic_funcs.py index ad51855..942aa36 100644 --- a/src/pipelines/shared/extraction/dynamic_funcs.py +++ b/src/pipelines/shared/extraction/dynamic_funcs.py @@ -42,7 +42,7 @@ def dynamic_primary( dynamic_reimbursement_fields = FieldSet() exhibit_level_answer_dict = {} - for field in dynamic_primary_fields.fields: + for field in list(dynamic_primary_fields.fields): exhibit_text_answer = prompt_calls.prompt_dynamic_primary( exhibit_text, field, @@ -230,25 +230,26 @@ def add_one_to_one_field( field_to_add.relationship = "one_to_one" # Special Handling for Program, Product, and Network: - # If any LOB value has been found, skip PROGRAM, PRODUCT, and NETWORK - # Only search for these fields when NO LOB has been found + # Skip when full LOB (all rows have LOB). When partial LOB, pass PROGRAM/PRODUCT; + # merge fills only empty cells, so 1:N values are preserved. if field_to_add.field_name in ["PROGRAM", "PRODUCT", "NETWORK"]: - # LOB can be a string or a list (from JSON format) - # Extract all LOB values, handling both string and list formats lob_values = [] for answer_dict in answer_dicts: lob_value = answer_dict.get("LOB", "") if isinstance(lob_value, list): - # If it's a list, extract each item and ensure it's a string for item in lob_value: if not string_utils.is_empty(item): lob_values.append(str(item)) elif not string_utils.is_empty(lob_value): - # If it's a string, add it directly (already hashable) lob_values.append(lob_value) - unique_lobs = set(lob_values) - if len(unique_lobs) > 0: + has_lob = len(set(lob_values)) > 0 + lob_empty_count = sum( + 1 for d in answer_dicts if string_utils.is_empty(d.get("LOB")) + ) + partial_lob = has_lob and lob_empty_count > 0 + # Skip only when full LOB; allow PROGRAM, PRODUCT when partial + if has_lob and not partial_lob: return one_to_one_fields if field_to_add.field_name == "CLAIM_TYPE_CD": @@ -313,6 +314,13 @@ def get_dynamic_one_to_one_fields( unique_lobs = set(lob_values) has_lob = len(unique_lobs) > 0 + # Partial LOB: some rows have LOB, some don't (e.g. stripped headers) + # When partial, pass LOB+PROGRAM+PRODUCT to 1:1; merge fills only empty cells + lob_empty_count = sum( + 1 for d in answer_dicts if string_utils.is_empty(d.get("LOB")) + ) + partial_lob = has_lob and lob_empty_count > 0 + # Handle ALL empty fields for field in all_empty_fields.fields: field_name = field.field_name @@ -325,11 +333,25 @@ def get_dynamic_one_to_one_fields( if string_utils.is_empty(answer_dict.get(field_name)) ) total_count = len(answer_dicts) + base = field.base_field if field.base_field else field_name - if empty_count == total_count: - # Skip PROGRAM, PRODUCT, NETWORK if any LOB has been found - # Only search for these fields when NO LOB has been found - if has_lob and field_name in ["PROGRAM", "PRODUCT", "NETWORK"]: + # LOB: pass when empty in ANY row (partial detection due to stripped headers) + # PROGRAM, PRODUCT: also pass when partial_lob and empty in any row (LOB relationship) + # Others: pass only when empty in ALL rows + should_pass_to_1to1 = ( + empty_count == total_count + or (base == "LOB" and empty_count > 0) + or (partial_lob and base in ["PROGRAM", "PRODUCT"] and empty_count > 0) + ) + + if should_pass_to_1to1: + # Skip PROGRAM, PRODUCT, NETWORK when full LOB (all rows have LOB) + # Do NOT skip when partial_lob; pass PROGRAM, PRODUCT for consistency + if ( + has_lob + and not partial_lob + and base in ["PROGRAM", "PRODUCT", "NETWORK"] + ): continue field_to_add = Field.load_from_file( @@ -339,8 +361,5 @@ def get_dynamic_one_to_one_fields( one_to_one_fields = add_one_to_one_field( one_to_one_fields, field_to_add, answer_dicts, constants ) - else: - # Field found in some rows - keep in 1:N - pass return one_to_one_fields diff --git a/src/pipelines/shared/extraction/one_to_n_funcs.py b/src/pipelines/shared/extraction/one_to_n_funcs.py index db0cea7..a692f69 100644 --- a/src/pipelines/shared/extraction/one_to_n_funcs.py +++ b/src/pipelines/shared/extraction/one_to_n_funcs.py @@ -935,6 +935,9 @@ def lesser_of_distribution( return final_reimbursement_level_answers +# TODO: DELETE - Outdated. Replaced by Exhibit-based prev_exhibit inheritance in +# file_processing.run_one_to_n_prompts. Uses get_previous_exhibit_dynamic_fields() on +# Exhibit; requires running exhibit_level for exhibits without reimbursements (see docs). def check_and_combine_exhibit_inheritance( previous_exhibit: dict | None, current_exhibit_page_nums: list[str],