Merged in feature/claim-type-only-runner (pull request #876)

Feature/claim type only runner * Add specific_fields config for running extraction on field groups ## What Changed 4 files modified: 1. src/config.py - Added configuration for field-specific extraction: - SPECIFIC_FIELDS arg (default: 'all') - pass field group name or comma-separated field names - FIELD_GROUPS dict - predefined groups: claim_type, dates, provider - get_specific_fields_list() - resolves config to actual field list 2. src/prompts/fieldset.py - Added filter_by_names() method to FieldSet class to filter fields by a list of names 3. src/pipelines/shared/extraction/one_to_n_funcs.py - Updated exhibit_level() to accept specific_fields parameter and skip prompts for fields not in the list 4. src/pipelines/saas/file_processing.py - Passes specific_fields through the call chain to both one_to_n and one_to_one extraction ## How It Works When specific_fields is set to something other than 'all': 1. Config resolves the field list (either from FIELD_GROUPS dict or comma-separated names) 2. Before running extraction prompt… * Add specific_fields filtering for claim_type only runs and fix UTF-8 logging * Merge remote-tracking branch 'origin/DEV' into feature/claim-type-only-runner * Fix black formatting in one_to_n_funcs.py * Fix black formatting * Fix CLAIM_TYPE_CD and AARETE_DERIVED_CLAIM_TYPE_CD to be single values - Handle pipe-delimited strings in crosswalk by splitting them - For CLAIM_TYPE_CD, take only first value in crosswalk since claim type should be singular - Add single_value_fields handling in normalize_field_value for CLAIM_TYPE_CD fields - Ensures AARETE_DERIVED_CLAIM_TYPE_CD is always a single M or H value * Merged in feature/fix-claim-type-mapping (pull request #878) Feature/fix claim type mapping * Fix CLAIM_TYPE_CD and AARETE_DERIVED_CLAIM_TYPE_CD to be single values - Handle pipe-delimited strings in crosswalk by splitting them - For CLAIM_TYPE_CD, take only first value in crosswalk since claim type should be singular - Add single_value_fields handling in normalize_field_value for CLAIM_TYPE_CD fields - Ensures AARETE_DERIVED_CLAIM_TYPE_CD is always a single M or H value * Update prompt template * Merged feature/claim-type-only-runner into feature/fix-claim-type-mapping Approved-by: Faizan Mohiuddin * Address PR review comments - Remove single_value_fields special handling from formatting_utils.py - Remove CLAIM_TYPE_CD only section and pipe-delimited handling from aarete_derived.py - Reference config.FIELD_GROUPS for provider_fields and date_fields in file_processing.py * Fix CLAIM_TYPE_CD to AARETE_DERIVED_CLAIM_TYPE_CD mapping for claim_type only runs - Apply crosswalk mapping in skip_reimbursement_processing block so derived fields are correctly mapped even when running claim_type only extraction - Fix _normalize_to_str to return first value when list has multiple elements instead of returning the list itself (violated str return type) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * normalize list to str-list Approved-by: Katon Minhas
2026-02-11 22:04:21 +00:00
parent 6fba0e7574
commit 93547e7b0d
7 changed files with 265 additions and 101 deletions
@@ -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"
@@ -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:
@@ -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
@@ -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)
@@ -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(
@@ -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)
@@ -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)