diff --git a/src/config.py b/src/config.py index 180807d..9910ef0 100644 --- a/src/config.py +++ b/src/config.py @@ -108,6 +108,48 @@ ENABLE_RUNTIME_ROTATION = False # Set to True to enable multi-runtime failover FIELDS = get_arg_value("fields", "all") # Valid: one_to_one, one_to_n, all FIELD_JSON_PATH = "src/prompts/investment_prompts.json" +# Specific field or field group to extract (comma-separated) +# Valid: "all", "claim_type", or specific field names like "CLAIM_TYPE_CD,CONTRACT_TITLE" +# "claim_type" is a shortcut for CLAIM_TYPE_CD + AARETE_DERIVED_CLAIM_TYPE_CD + CONTRACT_TITLE +SPECIFIC_FIELDS = get_arg_value("specific_fields", "all") + +# Field groups - predefined sets of related fields +FIELD_GROUPS = { + "claim_type": [ + "CLAIM_TYPE_CD", + "AARETE_DERIVED_CLAIM_TYPE_CD", + "CONTRACT_TITLE", + ], + "dates": [ + "EFFECTIVE_DT", + "AARETE_DERIVED_EFFECTIVE_DT", + "TERMINATION_DT", + "AARETE_DERIVED_TERMINATION_DT", + ], + "provider": [ + "PROV_GROUP_TIN", + "PROV_GROUP_NPI", + "PROV_GROUP_NAME_FULL", + "PROV_OTHER_TIN", + "PROV_OTHER_NPI", + "PROV_OTHER_NAME_FULL", + ], +} + + +def get_specific_fields_list(): + """Get the list of specific fields to extract based on config.""" + if SPECIFIC_FIELDS == "all": + return None # None means extract all fields + + # Check if it's a field group + if SPECIFIC_FIELDS in FIELD_GROUPS: + return FIELD_GROUPS[SPECIFIC_FIELDS] + + # Otherwise, treat as comma-separated list of field names + return [f.strip() for f in SPECIFIC_FIELDS.split(",") if f.strip()] + + # Client-specific prompts CLOVER_PROMPTS_PATH = "src/prompts/clover_prompts.json" BCBS_PROMISE_PROMPTS_PATH = "src/prompts/bcbs_promise_prompts.json" diff --git a/src/pipelines/saas/file_processing.py b/src/pipelines/saas/file_processing.py index 7aec330..9ade81a 100644 --- a/src/pipelines/saas/file_processing.py +++ b/src/pipelines/saas/file_processing.py @@ -86,6 +86,8 @@ def process_file(file_object, constants: Constants, run_timestamp): f"{datetime_str()} Starting One-to-N extraction for {len(exhibit_chunk_mapping)} exhibits - {filename}" ) with timing_utils.timed_block("one_to_n_extraction", context=filename): + # Get specific fields to extract (if configured) + specific_fields = config.get_specific_fields_list() ( one_to_n_results, dynamic_one_to_one_fields, @@ -97,6 +99,7 @@ def process_file(file_object, constants: Constants, run_timestamp): all_exhibit_headers=all_exhibit_headers, constants=constants, filename=filename, + specific_fields=specific_fields, ) if not one_to_n_results.empty: one_to_n_results["FILE_NAME"] = filename @@ -122,6 +125,8 @@ def process_file(file_object, constants: Constants, run_timestamp): # ONE TO ONE PROCESSING if process_one_to_one: + # Get specific fields to extract (if configured) + specific_fields = config.get_specific_fields_list() with timing_utils.timed_block("one_to_one_extraction", context=filename): one_to_one_results = run_one_to_one_prompts( filename, @@ -131,6 +136,7 @@ def process_file(file_object, constants: Constants, run_timestamp): dynamic_one_to_one_fields, first_reimbursement_page, constants, + specific_fields=specific_fields, ) one_to_one_results["FILE_NAME"] = filename logging.info(f"{datetime_str()} One to One Complete - {filename}") @@ -184,6 +190,7 @@ def run_one_to_one_prompts( dynamic_one_to_one_fields: FieldSet, first_reimbursement_page: str, constants: Constants, + specific_fields: Optional[list] = None, ): ################## INITIALIZE FIELDS ################## @@ -191,31 +198,51 @@ def run_one_to_one_prompts( relationship="one_to_one", file_path=config.FIELD_JSON_PATH ).combine(dynamic_one_to_one_fields) + # Filter fields if specific_fields is provided + if specific_fields is not None: + one_to_one_fields = one_to_one_fields.filter_by_names(specific_fields) + + # Check if we need provider info fields + provider_fields = set(config.FIELD_GROUPS.get("provider", [])) + run_provider_info = specific_fields is None or any( + f in provider_fields for f in specific_fields + ) + ################## RUN PROVIDER INFO ################## - with timing_utils.timed_block("one_to_one.provider_info", context=filename): - one_to_one_results = tin_npi_funcs.run_provider_info_fields( - contract_text, text_dict, filename, payer_name="" - ) + one_to_one_results = {} + if run_provider_info: + with timing_utils.timed_block("one_to_one.provider_info", context=filename): + one_to_one_results = tin_npi_funcs.run_provider_info_fields( + contract_text, text_dict, filename, payer_name="" + ) ################## RUN HYBRID SMART CHUNKED PROMPTS ################## # RAG function loads retrieval questions internally and matches with investment_prompts.json - with timing_utils.timed_block("one_to_one.hybrid_smart_chunking", context=filename): - hybrid_smart_chunked_answers_dict = ( - hybrid_smart_chunking_funcs.run_hybrid_smart_chunked_fields( - one_to_one_fields, constants, contract_text, filename, text_dict + # Only run if there are fields to extract + hybrid_smart_chunked_answers_dict = {} + if one_to_one_fields.contains_fields(): + with timing_utils.timed_block( + "one_to_one.hybrid_smart_chunking", context=filename + ): + hybrid_smart_chunked_answers_dict = ( + hybrid_smart_chunking_funcs.run_hybrid_smart_chunked_fields( + one_to_one_fields, constants, contract_text, filename, text_dict + ) ) - ) ################## RUN FULL CONTEXT PROMPTS ################## - with timing_utils.timed_block("one_to_one.full_context", context=filename): - full_context_answers_dict = one_to_one_funcs.run_full_context_fields( - one_to_one_fields, - contract_text, - text_dict, - first_reimbursement_page, - constants, - filename, - ) + # Only run if there are fields to extract + full_context_answers_dict = {} + if one_to_one_fields.contains_fields(): + with timing_utils.timed_block("one_to_one.full_context", context=filename): + full_context_answers_dict = one_to_one_funcs.run_full_context_fields( + one_to_one_fields, + contract_text, + text_dict, + first_reimbursement_page, + constants, + filename, + ) ################## COMBINE ANSWERS ################ # Prefer HSC answers over full_context when HSC value is not N/A for key, value in full_context_answers_dict.items(): @@ -228,19 +255,25 @@ def run_one_to_one_prompts( # Add HSC's N/A if field doesn't exist yet one_to_one_results[key] = value - one_to_one_results = tin_npi_funcs.merge_provider_info_with_one_to_one( - one_to_one_results, filename - ) + if run_provider_info: + one_to_one_results = tin_npi_funcs.merge_provider_info_with_one_to_one( + one_to_one_results, filename + ) ################## ADD AD FIELDS ################ - one_to_one_results = one_to_one_funcs.get_aarete_derived_dates( - one_to_one_results, text_dict, filename - ) + # Only run if date fields are requested + date_fields = set(config.FIELD_GROUPS.get("dates", [])) + if specific_fields is None or any(f in date_fields for f in specific_fields): + one_to_one_results = one_to_one_funcs.get_aarete_derived_dates( + one_to_one_results, text_dict, filename + ) ################## ADD SIGNATURE COUNT ################## - one_to_one_results = one_to_one_funcs.add_signature_count( - one_to_one_results, contract_text - ) + # Only run if signature count is requested + if specific_fields is None or "NUM_SIGNED_SIGNATORY_LINE_COUNT" in specific_fields: + one_to_one_results = one_to_one_funcs.add_signature_count( + one_to_one_results, contract_text + ) ################## Crosswalk Fields ################## # All field format normalization is handled at prompt_calls level via field-aware parsers @@ -264,6 +297,7 @@ def process_page_reimbursements( exhibit_page_nums: Optional[list[str]] = None, constants: Optional[Constants] = None, filename: Optional[str] = None, + specific_fields: Optional[list] = None, ): """ STEP 1 HELPER: Extract reimbursements from a single page (without exhibit-level fields). @@ -277,6 +311,9 @@ def process_page_reimbursements( exhibit_page_nums: List of page numbers in the exhibit (for backward compatibility) constants: Constants object filename: Name of the file being processed + specific_fields: Optional list of specific field names to extract. + If provided with exhibit-level-only fields (like claim_type), + skip detailed reimbursement extraction. """ # Get page text - prefer using exhibit.get_page_text() if available if exhibit.pages is not None: @@ -325,6 +362,18 @@ def process_page_reimbursements( if not reimbursement_level_answers: return [], [] + # If specific_fields is set and doesn't include reimbursement-level fields, + # skip detailed extraction (carveouts, lesser_of, breakout) - just return + # minimal info to trigger exhibit_level processing + if specific_fields is not None: + # Check if any reimbursement-level fields are requested + reimbursement_fields = {"SERVICE_TERM", "REIMB_TERM", "REIMB_PAGE"} + if not any(f in reimbursement_fields for f in specific_fields): + # Just return minimal reimbursement info to indicate this page has reimbursements + for answer_dict in reimbursement_level_answers: + answer_dict["REIMB_PAGE"] = page_num + return reimbursement_level_answers, [] + ################################ Carveouts and Special Case ################################ reimbursement_level_answers, special_case_answers = ( one_to_n_funcs.carveout_and_special_case( @@ -365,6 +414,7 @@ def run_one_to_n_prompts( all_exhibit_headers: Optional[dict[str, str]] = None, constants: Optional[Constants] = None, filename: Optional[str] = None, + specific_fields: Optional[list] = None, ): """ 3-STEP APPROACH (per exhibit): @@ -379,6 +429,8 @@ def run_one_to_n_prompts( all_exhibit_headers: Dictionary mapping exhibit pages to their headers constants: Constants object filename: Name of the file being processed + specific_fields: Optional list of specific field names to extract. + If provided, only these fields will be extracted at exhibit level. """ if exhibit_chunk_mapping is None: exhibit_chunk_mapping = {} @@ -420,6 +472,7 @@ def run_one_to_n_prompts( exhibit_pages, constants, filename, + specific_fields, ): page_num for page_num in exhibit_pages } @@ -450,6 +503,7 @@ def run_one_to_n_prompts( exhibit.exhibit_page, constants, filename, + specific_fields=specific_fields, ) ) exhibit.set_exhibit_level_data( @@ -459,49 +513,68 @@ def run_one_to_n_prompts( f"Exhibit Level Answers for {filename}, Page {exhibit.exhibit_page}: {exhibit_level_answers}" ) - # STEP 3: dynamic assignment & combine for this exhibit - # Get dynamic fields from previous exhibit if needed (for future use) - if exhibit.dynamic_reimbursement_fields: - dynamic_fields = exhibit.dynamic_reimbursement_fields - elif ( - exhibit.prev_exhibit - and not exhibit.prev_exhibit.has_reimbursements - and exhibit.get_previous_exhibit_dynamic_fields().fields - ): - dynamic_fields = exhibit.get_previous_exhibit_dynamic_fields() - else: - dynamic_fields = None + # Check if we need reimbursement-level processing + # If specific_fields is set and doesn't include reimbursement fields, skip Step 3 + reimbursement_fields = {"SERVICE_TERM", "REIMB_TERM", "REIMB_PAGE"} + skip_reimbursement_processing = specific_fields is not None and not any( + f in reimbursement_fields for f in specific_fields + ) - # Run dynamic assignment for ALL reimbursement rows in this exhibit - # Note: dynamic_assignment expects a list of rows and returns a list of rows - reimbursement_rows_with_dynamic = exhibit.reimbursement_rows - if exhibit.reimbursement_rows and dynamic_fields is not None: - reimbursement_rows_with_dynamic = dynamic_funcs.dynamic_assignment( - exhibit.reimbursement_rows, - dynamic_fields, - pages_dict, - text_dict, - exhibit, - constants, + if skip_reimbursement_processing: + # For exhibit-level-only extraction (like claim_type), just create minimal rows + # with exhibit_level_answers to preserve the data + if exhibit.exhibit_level_answers: + minimal_row = exhibit.exhibit_level_answers.copy() + minimal_row["REIMB_PAGE"] = exhibit.exhibit_page + # Apply crosswalk mapping for derived fields (e.g., CLAIM_TYPE_CD -> AARETE_DERIVED_CLAIM_TYPE_CD) + minimal_rows_with_crosswalk = aarete_derived.get_crosswalk_fields( + [minimal_row], constants + ) + one_to_n_results.extend(minimal_rows_with_crosswalk) + else: + # STEP 3: dynamic assignment & combine for this exhibit + # Get dynamic fields from previous exhibit if needed (for future use) + if exhibit.dynamic_reimbursement_fields: + dynamic_fields = exhibit.dynamic_reimbursement_fields + elif ( + exhibit.prev_exhibit + and not exhibit.prev_exhibit.has_reimbursements + and exhibit.get_previous_exhibit_dynamic_fields().fields + ): + dynamic_fields = exhibit.get_previous_exhibit_dynamic_fields() + else: + dynamic_fields = None + + # Run dynamic assignment for ALL reimbursement rows in this exhibit + # Note: dynamic_assignment expects a list of rows and returns a list of rows + reimbursement_rows_with_dynamic = exhibit.reimbursement_rows + if exhibit.reimbursement_rows and dynamic_fields is not None: + reimbursement_rows_with_dynamic = dynamic_funcs.dynamic_assignment( + exhibit.reimbursement_rows, + dynamic_fields, + pages_dict, + text_dict, + exhibit, + constants, + filename, + ) + + # Combine reimbursement rows with exhibit-level answers + combined_rows = row_funcs.combine_one_to_n( + exhibit.exhibit_text, + reimbursement_rows_with_dynamic, + exhibit.special_case_rows, + exhibit.exhibit_level_answers, filename, ) - # Combine reimbursement rows with exhibit-level answers - combined_rows = row_funcs.combine_one_to_n( - exhibit.exhibit_text, - reimbursement_rows_with_dynamic, - exhibit.special_case_rows, - exhibit.exhibit_level_answers, - filename, - ) + # Run cleaning + combined_rows = one_to_n_funcs.one_to_n_cleaning( + combined_rows, exhibit.exhibit_text, constants, filename + ) - # Run cleaning - combined_rows = one_to_n_funcs.one_to_n_cleaning( - combined_rows, exhibit.exhibit_text, constants, filename - ) - - exhibit.final_rows = combined_rows - one_to_n_results.extend(combined_rows) + exhibit.final_rows = combined_rows + one_to_n_results.extend(combined_rows) # Track first reimbursement page if first_reimbursement_page == "1" and exhibit.has_reimbursements: diff --git a/src/pipelines/shared/extraction/one_to_n_funcs.py b/src/pipelines/shared/extraction/one_to_n_funcs.py index 56c129a..db0cea7 100644 --- a/src/pipelines/shared/extraction/one_to_n_funcs.py +++ b/src/pipelines/shared/extraction/one_to_n_funcs.py @@ -18,6 +18,7 @@ def exhibit_level( exhibit_page: str, constants: Constants, filename: str, + specific_fields: list = None, ) -> tuple[dict, FieldSet]: """ Processes exhibit-level fields, including dynamic fields. @@ -28,6 +29,8 @@ def exhibit_level( reimbursement_level_fields (FieldSet): A FieldSet object containing reimbursement-level field prompts. constants (Constants): A Constants object filename (str): The name of the file + specific_fields (list): Optional list of specific field names to extract. + If provided, only these fields will be extracted. Returns: dict: All Exhibit-Level answers FieldSet: Updated reimbursement_level_fields with any dynamically added fields @@ -41,41 +44,62 @@ def exhibit_level( ) dynamic_code_fields = FieldSet(config.FIELD_JSON_PATH, field_type="dynamic_code") + # Filter fields if specific_fields is provided + if specific_fields is not None: + exhibit_level_fields = exhibit_level_fields.filter_by_names(specific_fields) + dynamic_primary_fields = dynamic_primary_fields.filter_by_names(specific_fields) + dynamic_reimb_info_fields = dynamic_reimb_info_fields.filter_by_names( + specific_fields + ) + dynamic_code_fields = dynamic_code_fields.filter_by_names(specific_fields) + # True Exhibit Level - these apply to the entire Exhibit - exhibit_level_answers = prompt_calls.prompt_exhibit_level( - exhibit_text, exhibit_level_fields, constants, filename - ) + exhibit_level_answers = {} + if exhibit_level_fields.contains_fields(): + exhibit_level_answers = prompt_calls.prompt_exhibit_level( + exhibit_text, exhibit_level_fields, constants, filename + ) - exhibit_level_answers = prompt_calls.prompt_exhibit_level_breakout( - exhibit_level_answers, filename - ) + exhibit_level_answers = prompt_calls.prompt_exhibit_level_breakout( + exhibit_level_answers, filename + ) - # Dynamic Primary - exhibit_level_answers, dynamic_reimbursement_fields = dynamic_funcs.dynamic_primary( - exhibit_text, exhibit_level_answers, constants, filename, dynamic_primary_fields - ) + # Dynamic Primary (skip if filtering to specific fields that don't include dynamic_primary) + dynamic_reimbursement_fields = FieldSet() + if dynamic_primary_fields.contains_fields(): + exhibit_level_answers, dynamic_reimbursement_fields = ( + dynamic_funcs.dynamic_primary( + exhibit_text, + exhibit_level_answers, + constants, + filename, + dynamic_primary_fields, + ) + ) - # Dynamic Codes - exhibit_level_answers, dynamic_reimbursement_fields = dynamic_funcs.dynamic( - exhibit_text, - exhibit_header, - exhibit_level_answers, - constants, - filename, - dynamic_code_fields, - dynamic_reimbursement_fields, - ) + # Dynamic Codes (skip if filtering to specific fields) + if dynamic_code_fields.contains_fields(): + exhibit_level_answers, dynamic_reimbursement_fields = dynamic_funcs.dynamic( + exhibit_text, + exhibit_header, + exhibit_level_answers, + constants, + filename, + dynamic_code_fields, + dynamic_reimbursement_fields, + ) - # Dynamic Reimb Info - exhibit_level_answers, dynamic_reimbursement_fields = dynamic_funcs.dynamic( - exhibit_text, - exhibit_header, - exhibit_level_answers, - constants, - filename, - dynamic_reimb_info_fields, - dynamic_reimbursement_fields, - ) + # Dynamic Reimb Info (skip if filtering to specific fields) + if dynamic_reimb_info_fields.contains_fields(): + exhibit_level_answers, dynamic_reimbursement_fields = dynamic_funcs.dynamic( + exhibit_text, + exhibit_header, + exhibit_level_answers, + constants, + filename, + dynamic_reimb_info_fields, + dynamic_reimbursement_fields, + ) # add exhibit identifiers exhibit_level_answers["EXHIBIT_PAGE"] = exhibit_page diff --git a/src/prompts/fieldset.py b/src/prompts/fieldset.py index 5f9d21c..e3373b5 100644 --- a/src/prompts/fieldset.py +++ b/src/prompts/fieldset.py @@ -378,3 +378,21 @@ class FieldSet: if field.field_name not in self.list_fields(): combined_fields.append(field) return FieldSet(fields=combined_fields) + + def filter_by_names(self, field_names: list): + """ + Filter fields to only include those in the provided list of field names. + + Args: + field_names: List of field names to keep. If None, returns all fields. + + Returns: + A new FieldSet containing only fields with names in the provided list. + """ + if field_names is None: + return FieldSet(fields=self.fields.copy()) + + filtered_fields = [ + field for field in self.fields if field.field_name in field_names + ] + return FieldSet(fields=filtered_fields) diff --git a/src/prompts/prompt_templates.py b/src/prompts/prompt_templates.py index 3c057c4..01eebc4 100644 --- a/src/prompts/prompt_templates.py +++ b/src/prompts/prompt_templates.py @@ -253,14 +253,14 @@ def EXHIBIT_LEVEL_INSTRUCTION() -> str: Extract attributes from a section of a contract. [GENERAL INSTRUCTIONS] -- For each field, return ALL correct answers found. There may be more than one correct answer. If there are multiple, return them as a JSON list within the dictionary value (e.g., "field_name": ["value1", "value2"]). +- For each field, return the correct answer. There will only be one correct answer. - If a list of valid values is provided, ONLY answer with the EXACT value from that list. - If there are no correct answers for a given field, write 'N/A' as the answer for that field. - Be consistent and deterministic; avoid creative paraphrasing. [OUTPUT FORMAT] Briefly explain your answer, then put the final answer in the properly-formatted JSON dictionary. -{JSON_DICT_WITH_LISTS_FORMAT_INSTRUCTIONS}""" +{JSON_DICT_FORMAT_INSTRUCTIONS}""" def EXHIBIT_LEVEL( diff --git a/src/utils/formatting_utils.py b/src/utils/formatting_utils.py index c2ee6fc..e4a030b 100644 --- a/src/utils/formatting_utils.py +++ b/src/utils/formatting_utils.py @@ -81,8 +81,8 @@ def _normalize_to_str(value: Any) -> str: 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) + # Multiple values + return str(value) elif isinstance(value, dict): # Convert dict to string representation return json.dumps(value) diff --git a/src/utils/logging_utils.py b/src/utils/logging_utils.py index cb367ba..9d638b5 100644 --- a/src/utils/logging_utils.py +++ b/src/utils/logging_utils.py @@ -28,7 +28,9 @@ Architecture: """ import contextvars +import io import logging +import sys from pathlib import Path from src import config @@ -96,7 +98,8 @@ class PerFileHandler(logging.Handler): # Get cached handler or create a new one if not exists if safe_filename not in self._file_handlers: # mode="w" to overwrite existing per-file logs for fresh run - file_handler = logging.FileHandler(log_file, mode="w") + # encoding="utf-8" to handle Unicode characters (arrows, special chars) + file_handler = logging.FileHandler(log_file, mode="w", encoding="utf-8") formatter = logging.Formatter( "%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) @@ -193,7 +196,11 @@ def setup_per_file_logging() -> None: ) # Always add console output for monitoring overall progress - console_handler = logging.StreamHandler() + # Wrap stdout with UTF-8 encoding to handle Unicode characters (arrows, etc.) on Windows + utf8_stdout = io.TextIOWrapper( + sys.stdout.buffer, encoding="utf-8", errors="replace" + ) + console_handler = logging.StreamHandler(utf8_stdout) console_handler.setLevel(logging.INFO) console_formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") console_handler.setFormatter(console_formatter)