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,