From 53a0c76d8ec2e2bb3a3958995413f7a152ad68c2 Mon Sep 17 00:00:00 2001 From: Praneel Panchigar Date: Tue, 19 May 2026 14:37:23 +0000 Subject: [PATCH] Merged in bugfix/stg-to-main-prep (pull request #1012) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bugfix/stg to main prep * Merged in dev (pull request #1001) Dev * Revert premature merge of bugfix/retire_stale_client_file_processing PR #959 was merged into dev without approval. This reverts commits 5e143c10, 63f32c41, 849aa626, and 927abcae to restore dev to its pre-merge state. The changes will be re-submitted via a new PR after proper review. * Merged in bugfix/retire_stale_client_file_processing (pull request #961) Return None for dashboard output when dashboard postprocessing is off * Return None for dashboard output when dashboard postprocessing is off FINAL_RESULT_DF_DASHBOARD was initialized as an empty DataFrame even when RUN_DASHBOARD_POSTPROCESSING was False, causing downstream code to needlessly process it (reorder_columns, etc). Now returns None when dashboard is not requested, matching the postprocess() contract. * Merged dev into bugfix/retire_stale_client_file_processing * Merge dev (with revert) into feature branch * Re-apply retire stale client file_processing changes Revert of the revert (4f53b528) to… * fix grouping_key casing in service term standardization qc.py uppercases all PC_Details columns to GROUPING_KEY, but the service-term reader looked up lowercase 'grouping_key'. The lookup silently failed and degraded per-group standardization to global mode. * enable prompt caching on service_term_standardization LLM calls cache=True was missing; the 2.5k-token instruction was being re-billed on every chunk and mapping-batch call instead of reading from the shared instruction cache. * register SERVICE_ENRICHMENT_INSTRUCTION in cache registry The runtime call at code_funcs.py:75 already passes cache=True with usage_label="SERVICE_ENRICHMENT", but the registry entry was missing, so the prompt was never warmed and every first call paid create-tokens. * remove stray debug prints from document_classification.generate_pre_doczy_report Three print() calls were dumping the Program/Product mapping to stdout on every run. * demote code_explicit raw-answer log from INFO to DEBUG Full raw LLM responses were spamming INFO-level logs on every call. * flag vendor prompt_templates convention drift for follow-up Vendor pipeline still uses legacy hand-rolled JSON wording and has one stale 'Return result in pipes' reference. Adopting the JSON_*_FORMAT_INSTRUCTIONS macros from the main pipeline is deferred to a dedicated follow-up; header comment points to the review tracker. * skip healthcare finalizer for vendor pipelines runner.main applied add_aarete_derived_payer_name/provider_name and the FIELD_FORMAT_MAPPING reorder unconditionally to every client. The mapping is healthcare-only (src/constants/investment_columns.py), so reorder_columns dropped every vendor-specific column (TITLE, VENDOR_NAME, AMOUNT, SLA/KPI, ...) from the combined cc_results_full output. Per-file individual outputs survived because they are written inside generic_processor before the finalizer runs. Guard the entire healthcare finalizer block with registry.is_vendor(client) so vendor results keep their own flat schema. * move DTC short-circuit above duplicate_detection in runner In DTC mode, dedup ran twice: once in runner before the short-circuit (immediately discarded at return), and again inside dtc_main.main on the same input_dict. Reorder so DTC exits before the runner-scope dedup is computed. dtc_main keeps its own internal dedup pass. Inline comment notes the cleaner follow-up — have dtc_main.main take (original_files_to_process, file_status_map) so both call sites share a single computation. * remove dead code from saas/main.py This module is a thin delegator to runner.main; the actual safe_process_file lives in runner.py:101. The local copy at saas/main.py referenced logging, traceback, pd, and file_processing — none of which were imported, so any caller would have raised NameError immediately. Drops the function and the four unused module-level imports (parent_child_main, dtc_main, standardize_service_terms, qc_qa.pipeline.*). File shrinks from 102 to 28 lines. * Merge origin/dev into bugfix/stg-to-main-prep Reconciles the 9 stg-to-main prep fixes with 197 commits dev gained since 2026-03-16. Resolved 14 conflicts: Group A (preserve our 9 fixes): - saas/main.py: dead-code cleanup preserved - code_funcs.py: code_explicit log demoted to DEBUG preserved - document_classification/main.py: 3 stray prints removed preserved - vendors/prompt_templates.py: NOTE flagging legacy JSON wording preserved - cache_registry.py: SERVICE_ENRICHMENT registry entry preserved - service_term_standardization.py: GROUPING_KEY casing + cache=True preserved - runner.py: vendor finalizer guard + DTC short-circuit reorder preserved Group B (dev prevails; our fixes don't touch these): - saas/file_processing.py: dev's per-field confidence scoring (DAIP2-2692) - postprocessing_funcs.py: dev's add_amendment_intent + _CONF carve-out - preprocess.py: dev's removal of stale fallback comment - preprocessing_funcs.py: dev's _find_header_in_text import (needed by body) - prompts/prompt_templates.py: dev… Approved-by: Siddhant Medar --- src/codes/code_funcs.py | 2 +- src/document_classification/main.py | 3 - src/pipelines/runner.py | 87 +++++++++++-------- src/pipelines/saas/main.py | 73 ---------------- .../service_term_standardization.py | 3 +- src/pipelines/vendors/prompt_templates.py | 8 ++ src/prompts/cache_registry.py | 6 ++ 7 files changed, 70 insertions(+), 112 deletions(-) diff --git a/src/codes/code_funcs.py b/src/codes/code_funcs.py index c622d31..a3f1719 100644 --- a/src/codes/code_funcs.py +++ b/src/codes/code_funcs.py @@ -113,7 +113,7 @@ def code_explicit(service: str, methodology: str, filename: str) -> dict: instruction=prompt_templates.CODE_EXPLICIT_INSTRUCTION(), usage_label="CODE_EXPLICIT", ) - logging.info("LLM raw answer for code_explicit: %s", llm_answer_raw) + logging.debug("LLM raw answer for code_explicit: %s", llm_answer_raw) try: code_answer_dict = _parser(llm_answer_raw) return code_answer_dict diff --git a/src/document_classification/main.py b/src/document_classification/main.py index 0199837..31275f6 100644 --- a/src/document_classification/main.py +++ b/src/document_classification/main.py @@ -996,9 +996,6 @@ def generate_pre_doczy_report(input_dict, run_timestamp): products_list = _mapping.valid_aarete_derived_products df["PROGRAM"] = [programs_list] * len(df) df["PRODUCT"] = [products_list] * len(df) - print("Program/Product mapping loaded for summary fields") - print(f"Sample PROGRAM values: {programs_list}") - print(f"Sample PRODUCT values: {products_list}") col_order = [ "FILE_NAME", "STATUS", diff --git a/src/pipelines/runner.py b/src/pipelines/runner.py index 73bf9b5..233a0f5 100644 --- a/src/pipelines/runner.py +++ b/src/pipelines/runner.py @@ -216,17 +216,17 @@ def main(client: str = "saas", testing=False, test_params={}): logging.info(f"Total Input Files : {len(input_dict)}") - # Detect and handle duplicate contracts - with timing_utils.timed_block("duplicate_detection", log_level="INFO"): - duplicate_groups = duplicate_detection.find_duplicate_groups(input_dict) - original_files_to_process, file_status_map = ( - duplicate_detection.get_files_to_process(input_dict, duplicate_groups) - ) - logging.debug( - f"duplicate_groups={len(duplicate_groups)}, original_files={len(original_files_to_process)}, file_status_map length={len(file_status_map)}" - ) - # ========== COMMON: Document Type Classification ========== + # DTC short-circuits before duplicate detection runs in this scope. + # dtc_main.main runs its own duplicate detection internally + # (document_classification/main.py:704), so computing it here too + # was a full second normalize+hash pass that got thrown away. + # + # Cleaner follow-up: have dtc_main.main accept + # (original_files_to_process, file_status_map) as parameters and + # share a single computation across both call sites — would let us + # move the duplicate_detection block back above this short-circuit + # without paying the cost twice. if config.PERFORM_DTC: logging.info("=" * 80) logging.info("DOCUMENT TYPE CLASSIFICATION MODE - Running DTC and exiting") @@ -238,6 +238,16 @@ def main(client: str = "saas", testing=False, test_params={}): logging.info("=" * 80) return + # Detect and handle duplicate contracts + with timing_utils.timed_block("duplicate_detection", log_level="INFO"): + duplicate_groups = duplicate_detection.find_duplicate_groups(input_dict) + original_files_to_process, file_status_map = ( + duplicate_detection.get_files_to_process(input_dict, duplicate_groups) + ) + logging.debug( + f"duplicate_groups={len(duplicate_groups)}, original_files={len(original_files_to_process)}, file_status_map length={len(file_status_map)}" + ) + # ========== COMMON: Warm prompt caches ========== # Drive warming from `cache_registry`, which knows the (usage_label, # runtime_model) pairing. Cache entries are model-specific in Bedrock, @@ -345,33 +355,42 @@ def main(client: str = "saas", testing=False, test_params={}): FINAL_RESULT_DF_CC = pd.DataFrame() FINAL_RESULT_DF_DASHBOARD = None - if not FINAL_RESULT_DF_CC.empty: - FINAL_RESULT_DF_CC = one_to_one_funcs.add_aarete_derived_payer_name( - FINAL_RESULT_DF_CC, config.STATE_FLAG - ) - FINAL_RESULT_DF_CC = one_to_one_funcs.add_aarete_derived_provider_name( - FINAL_RESULT_DF_CC, config.STATE_FLAG - ) - if ( - FINAL_RESULT_DF_DASHBOARD is not None - and not FINAL_RESULT_DF_DASHBOARD.empty - ): - FINAL_RESULT_DF_DASHBOARD = one_to_one_funcs.add_aarete_derived_payer_name( - FINAL_RESULT_DF_DASHBOARD, config.STATE_FLAG - ) - FINAL_RESULT_DF_DASHBOARD = ( - one_to_one_funcs.add_aarete_derived_provider_name( - FINAL_RESULT_DF_DASHBOARD, config.STATE_FLAG + # Healthcare-only finalizer: Aarete derivations and the + # FIELD_FORMAT_MAPPING reorder are tied to the healthcare schema + # (see src/constants/investment_columns.py). Vendor pipelines build + # their own flat schema in vendors/shared/generic_processor.py + # (TITLE, VENDOR_NAME, AMOUNT, SLA/KPI, ...), so running this block + # on vendor results would drop every vendor-specific column. + if not registry.is_vendor(client): + if not FINAL_RESULT_DF_CC.empty: + FINAL_RESULT_DF_CC = one_to_one_funcs.add_aarete_derived_payer_name( + FINAL_RESULT_DF_CC, config.STATE_FLAG + ) + FINAL_RESULT_DF_CC = one_to_one_funcs.add_aarete_derived_provider_name( + FINAL_RESULT_DF_CC, config.STATE_FLAG + ) + if ( + FINAL_RESULT_DF_DASHBOARD is not None + and not FINAL_RESULT_DF_DASHBOARD.empty + ): + FINAL_RESULT_DF_DASHBOARD = ( + one_to_one_funcs.add_aarete_derived_payer_name( + FINAL_RESULT_DF_DASHBOARD, config.STATE_FLAG + ) + ) + FINAL_RESULT_DF_DASHBOARD = ( + one_to_one_funcs.add_aarete_derived_provider_name( + FINAL_RESULT_DF_DASHBOARD, config.STATE_FLAG + ) ) - ) - FINAL_RESULT_DF_CC = postprocessing_funcs.reorder_columns( - FINAL_RESULT_DF_CC, FIELD_FORMAT_MAPPING - ) - if FINAL_RESULT_DF_DASHBOARD is not None: - FINAL_RESULT_DF_DASHBOARD = postprocessing_funcs.reorder_columns( - FINAL_RESULT_DF_DASHBOARD, FIELD_FORMAT_MAPPING + FINAL_RESULT_DF_CC = postprocessing_funcs.reorder_columns( + FINAL_RESULT_DF_CC, FIELD_FORMAT_MAPPING ) + if FINAL_RESULT_DF_DASHBOARD is not None: + FINAL_RESULT_DF_DASHBOARD = postprocessing_funcs.reorder_columns( + FINAL_RESULT_DF_DASHBOARD, FIELD_FORMAT_MAPPING + ) if len(error_results) > 0: ERROR_RESULT_DF = pd.concat(error_results, ignore_index=True) diff --git a/src/pipelines/saas/main.py b/src/pipelines/saas/main.py index c476458..13a202c 100644 --- a/src/pipelines/saas/main.py +++ b/src/pipelines/saas/main.py @@ -17,79 +17,6 @@ def _resolve_client() -> str: return "saas" -from src.parent_child import main as parent_child_main -from src.document_classification import main as dtc_main -from src.pipelines.shared.postprocessing.service_term_standardization import ( - standardize_service_terms, -) -from src.qc_qa.pipeline import ( - generate_statistics, - generate_tin_contract_summary, - generate_tin_statistics, - run_qc_qa_pipeline, - save_qc_qa_outputs, -) - - -def safe_process_file(item, constants, run_timestamp): - """Process a single file with comprehensive error handling. - - This function wraps the main file processing logic to ensure that: - - Individual file failures don't crash the entire batch - - Detailed error information is captured for debugging - - Args: - item: Tuple of (filename, contract_text) representing the file to process - constants: Constants object containing configuration and processing rules - run_timestamp: Timestamp string for output file naming and tracking - - Returns: - tuple: Either: - - (cc_df, dashboard_df) tuple with extracted field results (success case) - - (error_df, error_df) tuple with error details (failure case) - - Note: - Returns both CC and dashboard versions from file processing. - With FAISS migration, no special resource cleanup is needed. - """ - file_id = ( - item[0] if isinstance(item, (list, tuple)) and len(item) > 0 else item - ) # unpack file_id from item (passed in as a tuple below) - try: - cc_df, dashboard_df = file_processing.process_file( - item, constants, run_timestamp - ) - return cc_df, dashboard_df - except ( - Exception - ) as e: # When there's an issue with the processing inside the future - error_type = type(e).__name__ - error_message = str(e) - full_traceback = traceback.format_exc() - - logging.error(f"Error processing file {file_id}") - logging.error(f"Error type: {error_type}") - logging.error(f"Error Message: {error_message}") - - logging.error(f"Full traceback:\n{full_traceback}") - - error_df = pd.DataFrame( - [ - { - "FILE_NAME": file_id, - "error": error_message.replace(",", ";").replace( - "\n", " " - ), # Replace problematic characters for CSV compatibility - "error_type": error_type, - "traceback": full_traceback.replace(",", ";").replace( - "\n", " | " - ), # keep line breaks as separators - } - ] - ) # Return a single-row dataframe so we can still concat it later - return error_df, error_df # Return tuple for consistency - - def main(testing=False, test_params={}): from src.utils import instrumentation diff --git a/src/pipelines/shared/postprocessing/service_term_standardization.py b/src/pipelines/shared/postprocessing/service_term_standardization.py index 1fe4ce3..392fb28 100644 --- a/src/pipelines/shared/postprocessing/service_term_standardization.py +++ b/src/pipelines/shared/postprocessing/service_term_standardization.py @@ -58,7 +58,7 @@ _MAPPING_BATCH_SIZE = 100 # PC_Details constants (mirrors latest_contract_for_service.py) _PC_SHEET = "PC_Details" _PC_JOIN_KEY = "FILE_NAME" -_PC_GROUP_COL = "grouping_key" +_PC_GROUP_COL = "GROUPING_KEY" # ── Prompt functions (cacheable INSTRUCTION/PROMPT pattern) ─────────────────── @@ -230,6 +230,7 @@ def _call_llm(prompt: str) -> str: model_id=_MODEL_ID, filename="service_term_standardization", max_tokens=_MAX_TOKENS, + cache=True, instruction=SERVICE_TERM_STANDARDIZATION_INSTRUCTION(), usage_label="service_term_standardization", ) diff --git a/src/pipelines/vendors/prompt_templates.py b/src/pipelines/vendors/prompt_templates.py index 97078bb..e973b77 100644 --- a/src/pipelines/vendors/prompt_templates.py +++ b/src/pipelines/vendors/prompt_templates.py @@ -7,6 +7,14 @@ never affect non-vendor (SaaS/client) pipelines. Functions here mirror their counterparts in the main prompt_templates module but are owned by the vendor pipeline going forward. + +NOTE: this module currently uses the legacy hand-rolled JSON-format wording +(and one stale "Return result in pipes" reference). The main Doczy pipeline +in src/prompts/prompt_templates.py has standardized on the +JSON_DICT_FORMAT_INSTRUCTIONS / JSON_LIST_FORMAT_INSTRUCTIONS / etc. macros +interpolated at the end of each TEMPLATE. Vendor prompts should be migrated +to that convention in a follow-up; tracked in +.claude/plans/stg_to_main_review_20260513.md. """ # --------------------------------------------------------------------------- diff --git a/src/prompts/cache_registry.py b/src/prompts/cache_registry.py index 73605ee..b52b551 100644 --- a/src/prompts/cache_registry.py +++ b/src/prompts/cache_registry.py @@ -402,6 +402,12 @@ _REGISTRY: tuple[CachePromptEntry, ...] = ( cache_class=CacheClass.INSTRUCTION, ), # ---------- code extraction family ---------- + CachePromptEntry( + usage_label="SERVICE_ENRICHMENT", + instruction_func=prompt_templates.SERVICE_ENRICHMENT_INSTRUCTION, + runtime_model="sonnet_latest", + cache_class=CacheClass.INSTRUCTION, + ), CachePromptEntry( usage_label="CODE_EXPLICIT", instruction_func=prompt_templates.CODE_EXPLICIT_INSTRUCTION,