From a0c7e7a7386a4e088d5c661444e57ed000606052 Mon Sep 17 00:00:00 2001 From: Praneel Panchigar Date: Fri, 13 Feb 2026 21:44:46 +0000 Subject: [PATCH] Merged in bugfix/dynamic_issues_feb12 (pull request #882) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bugfix/dynamic issues feb12 * prompt changes reverted * fix pipeline issues * fix pipeline issues * fix pipeline issues * fixed formatting * fixed formatting * Merge branch 'DEV' into Optimize/DAIP2-1474-restructure-postprocess * Merge branch 'DEV' into Optimize/DAIP2-1474-restructure-postprocess * save dashboard and cc output separately * save dashboard output in s3 * pipeline error fixed * json list through postprocessing * Merged DEV into Optimize/DAIP2-1474-restructure-postprocess * Merge remote-tracking branch 'origin/Optimize/DAIP2-1474-restructure-postprocess' into feature/new_output_format * Restructure output file organization and add standard field sanitization Output Structure Changes: - Reorganize output files into hierarchical directory structure: - full_outputs/cc_results/ for consolidated CC results - full_outputs/dashboard_results/ for consolidated dashboard results - full_outputs/ for error files - automation_qa-qc/ for QC/QA validated results and statistics - parent-child/ for parent-child mapping outputs - tracking/ for usage and cost tracking data - individual/ for per-file CC results (dashboard individual files removed) - Update file naming conventions to match new structure - Remove QC/QA processing for error files (error files are saved without validation) Post-Processing Changes: - Add standard N/A value cleaning: remove placeholder values (N/A, UNKNOWN, etc.) when they are the only value in a cell (applies before CC/dashboard split) - Normalize all _IND fields to contain only 'Y' or 'N' values (no blanks) - Ensure standard cleaning runs before splitting into CC … * Ran Black for CI * Refactor file splitting logic and add comprehensive tests - Refactored splitting logic in io_utils.py: - Consolidated repeated splitting code into two focused helper functions: - _write_local_split_files() for local file writing with splitting - _write_s3_split_files() for S3 file writing with splitting - Both helpers use shared split_dataframe_by_filename() function - Added MAX_ROWS_PER_SPLIT configuration (default: 70000) in config.py - Added comprehensive test coverage for splitting logic: - Tests for split_dataframe_by_filename() with various scenarios - Tests for local and S3 write operations with single and multiple splits - Tests for cc_results_full, dashboard_results_full, and qc_qa_cc_full output types - Fixed existing test failures (write_s3 error handling, path assertions) - Improved code maintainability and readability * Fix failing tests in test_postprocess.py - Updated standard_postprocess tests to use actual columns from FIELD_FORMAT_MAPPING (PAYER_NAME, CONTRACT_TITLE) instead of custom test columns that get dropped - Added FILE_NAME column to all file structure test DataFrames (required for splitting logic) - Added MAX_ROWS_PER_SPLIT mock configuration for splitting tests - Fixed patch decorators for S3 tests to properly mock logging All 41 tests now passing. * Black for CI * Blank [] and ['[]'] in output instead of displaying them - Add placeholder patterns in clean_na_values for [], ['[]'], ["[]"] - Update format_as_json_list to return blank for empty lists instead of [] - Filter out empty-list placeholder items from list values in format_as_json_list - Add tests for clean_na_values empty list handling and format_as_json_list Co-authored-by: Cursor * Merged DEV into feature/new_output_format * feat: dynamic primary debug improvements and LOB partial 1:1 escalation - Add per-exhibit debug summary for dynamic primary discovery (page, values, raw LLM) - Pass exhibit_page to dynamic_primary for debug; mark debug-only params for removal - Fix 1:1 escalation skip: use base_field for PROGRAM/PRODUCT/NETWORK when LOB detected - Pass LOB to 1:1 when empty in any row (partial detection from stripped headers) - Add debug exhibit text preview; mark check_and_combine_exhibit_inheritance for deletion - prompt_dynamic_primary returns (answer, raw); remove per-field prints Co-authored-by: Cursor * feat: pass PROGRAM and PRODUCT to 1:1 when partial LOB When LOB is empty in some rows (partial detection), pass LOB, PROGRAM, and PRODUCT to 1:1 for contract-level extraction. Merge fills only empty cells so 1:N values are preserved. Skip PROGRAM/PRODUCT/NETWORK only when full LOB. Co-authored-by: Cursor * Removed debug print blocks * Merge origin/DEV into bugfix/dynamic_issues_feb12 Resolved conflicts: - runner.py: Use RUN_DASHBOARD_POSTPROCESSING for conditional dashboard; parent-child output_dir and S3 upload - main.py: Use RUN_DASHBOARD_POSTPROCESSING for conditional dashboard - postprocess.py: Optional dashboard postprocessing when RUN_DASHBOARD_POSTPROCESSING is True Co-authored-by: Cursor Approved-by: Katon Minhas --- src/pipelines/runner.py | 28 +++++----- src/pipelines/saas/prompts/prompt_calls.py | 20 +++++--- .../shared/extraction/dynamic_funcs.py | 51 +++++++++++++------ .../shared/extraction/one_to_n_funcs.py | 3 ++ 4 files changed, 64 insertions(+), 38 deletions(-) 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],