diff --git a/src/document_classification/main.py b/src/document_classification/main.py index 9c9f481..f028ca3 100644 --- a/src/document_classification/main.py +++ b/src/document_classification/main.py @@ -314,6 +314,10 @@ if __name__ == "__main__": level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", ) + + # Suppress AWS SDK logging + logging.getLogger("botocore").setLevel(logging.WARNING) + logging.getLogger("boto3").setLevel(logging.WARNING) # Generate timestamp for this run run_timestamp = f"run_{datetime.now().strftime('%Y%m%d_%H-%M')}_{config.BATCH_ID}" diff --git a/src/parent_child/qc.py b/src/parent_child/qc.py index 31388a9..9b49822 100644 --- a/src/parent_child/qc.py +++ b/src/parent_child/qc.py @@ -26,6 +26,10 @@ logging.basicConfig( ) logger = logging.getLogger(__name__) +# Suppress AWS SDK logging +logging.getLogger("botocore").setLevel(logging.WARNING) +logging.getLogger("boto3").setLevel(logging.WARNING) + # ============================================================ # Constants & Enums diff --git a/src/pipelines/clients/bcbs_promise/prompts/prompt_calls.py b/src/pipelines/clients/bcbs_promise/prompts/prompt_calls.py index 8af60ef..6ae42e3 100644 --- a/src/pipelines/clients/bcbs_promise/prompts/prompt_calls.py +++ b/src/pipelines/clients/bcbs_promise/prompts/prompt_calls.py @@ -4,7 +4,8 @@ import src.config as config import src.prompts.prompt_templates as prompt_templates from src.constants.constants import Constants from src.prompts.fieldset import Field, FieldSet -from src.utils import llm_utils, string_utils +from src.utils import llm_utils, string_utils, timing_utils, rag_utils +from src.pipelines.shared.preprocessing.hybrid_smart_chunking_funcs import format_chunks_for_llm import json @@ -852,3 +853,91 @@ def provider_name_match_check( ) return False +def prompt_hsc_single_field( + field, retriever, reranker, rag_config, text_dict, constants, filename +): + """Process a single field in parallel for HSC.""" + try: + retrieval_query = field.retrieval_question + + if not retrieval_query: + return (field.field_name, "N/A", field) + + # Retrieve + rerank using retrieval question + with timing_utils.timed_block( + f"hsc.field.{field.field_name}.retrieval", context=filename + ): + retrieved = retriever.invoke(retrieval_query)[: rag_config.ENSEMBLE_TOP_K] + reranked = rag_utils.rerank_documents( + retrieval_query, + retrieved, + reranker, + rag_config.RERANKER_TOP_K, + rag_config.RERANKER, + ) + + # Build context and use actual LLM prompt for extraction + context = format_chunks_for_llm(text_dict, reranked) + + if len(context) == 0: + # Fallback: mark field for full_context processing + field.field_type = "full_context" + logging.warning( + f"No context retrieved for {field.field_name}, marking as full_context" + ) + return (field.field_name, "N/A", field) + + llm_prompt = field.get_prompt_dict( + constants + ) # Returns dict of {field_name : prompt} + prompt, _parser = prompt_templates.ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(context, llm_prompt) + + logging.debug(f"Field: {field.field_name}") + logging.debug(f"Retrieval question: {retrieval_query}") + logging.debug(f"LLM prompt: {llm_prompt}") + + # LLM call + with timing_utils.timed_block( + f"hsc.field.{field.field_name}.llm_call", context=filename + ): + llm_answer_raw = llm_utils.invoke_claude( + prompt, "sonnet_latest", filename, max_tokens=8192 + ) + try: + llm_answer_final = _parser(llm_answer_raw) # returns dict + + val = llm_answer_final.get(field.field_name) + if isinstance(val, list): + llm_answer_final[field.field_name] = val + elif isinstance(val, str): + llm_answer_final[field.field_name] = [val] + + field_value = llm_answer_final[field.field_name] + if field_value[0] == "N/A": + field.field_type = "full_context" + if field.field_name == "PAYER_NAME": + field.prompt = ( + field.prompt + + prompt_templates.FULL_CONTEXT_PAYER_NAME_ADDITIONAL_INSTRUCTION() + ) + logging.warning( + f"Obtained N/A for {field.field_name}, marking as full_context" + ) + + return (field.field_name, field_value, field) + + except Exception as e: + logging.warning( + f"Failed to parse JSON response for field {field.field_name}: " + f"{type(e).__name__}: {str(e)}. Setting field value to N/A." + ) + return (field.field_name, "N/A", field) + + except Exception as e: + # Mark field for full_context retry and preserve error info + field.field_type = "full_context" + logging.error( + f"Error on {field.field_name}: {type(e).__name__}: {str(e)}, marking as full_context" + ) + return (field.field_name, "N/A", field) + diff --git a/src/pipelines/clients/clover/prompts/prompt_calls.py b/src/pipelines/clients/clover/prompts/prompt_calls.py index 8832169..5c141ee 100644 --- a/src/pipelines/clients/clover/prompts/prompt_calls.py +++ b/src/pipelines/clients/clover/prompts/prompt_calls.py @@ -3,9 +3,9 @@ import logging import src.config as config import src.prompts.prompt_templates as prompt_templates from src.constants.constants import Constants -from src.constants.delimiters import Delimiter from src.prompts.fieldset import Field, FieldSet -from src.utils import llm_utils, string_utils +from src.utils import llm_utils, string_utils, timing_utils, rag_utils +from src.pipelines.shared.preprocessing.hybrid_smart_chunking_funcs import format_chunks_for_llm import json @@ -907,3 +907,92 @@ def provider_name_match_check( f"[is_provider_name_match_llm] Error calling LLM: {e}, defaulting to False" ) return False + +def prompt_hsc_single_field( + field, retriever, reranker, rag_config, text_dict, constants, filename +): + """Process a single field in parallel for HSC.""" + try: + retrieval_query = field.retrieval_question + + if not retrieval_query: + return (field.field_name, "N/A", field) + + # Retrieve + rerank using retrieval question + with timing_utils.timed_block( + f"hsc.field.{field.field_name}.retrieval", context=filename + ): + retrieved = retriever.invoke(retrieval_query)[: rag_config.ENSEMBLE_TOP_K] + reranked = rag_utils.rerank_documents( + retrieval_query, + retrieved, + reranker, + rag_config.RERANKER_TOP_K, + rag_config.RERANKER, + ) + + # Build context and use actual LLM prompt for extraction + context = format_chunks_for_llm(text_dict, reranked) + + if len(context) == 0: + # Fallback: mark field for full_context processing + field.field_type = "full_context" + logging.warning( + f"No context retrieved for {field.field_name}, marking as full_context" + ) + return (field.field_name, "N/A", field) + + llm_prompt = field.get_prompt_dict( + constants + ) # Returns dict of {field_name : prompt} + prompt, _parser = prompt_templates.ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(context, llm_prompt) + + logging.debug(f"Field: {field.field_name}") + logging.debug(f"Retrieval question: {retrieval_query}") + logging.debug(f"LLM prompt: {llm_prompt}") + + # LLM call + with timing_utils.timed_block( + f"hsc.field.{field.field_name}.llm_call", context=filename + ): + llm_answer_raw = llm_utils.invoke_claude( + prompt, "sonnet_latest", filename, max_tokens=8192 + ) + try: + llm_answer_final = _parser(llm_answer_raw) # returns dict + + val = llm_answer_final.get(field.field_name) + if isinstance(val, list): + llm_answer_final[field.field_name] = val + elif isinstance(val, str): + llm_answer_final[field.field_name] = [val] + + field_value = llm_answer_final[field.field_name] + if field_value[0] == "N/A": + field.field_type = "full_context" + if field.field_name == "PAYER_NAME": + field.prompt = ( + field.prompt + + prompt_templates.FULL_CONTEXT_PAYER_NAME_ADDITIONAL_INSTRUCTION() + ) + logging.warning( + f"Obtained N/A for {field.field_name}, marking as full_context" + ) + + return (field.field_name, field_value, field) + + except Exception as e: + logging.warning( + f"Failed to parse JSON response for field {field.field_name}: " + f"{type(e).__name__}: {str(e)}. Setting field value to N/A." + ) + return (field.field_name, "N/A", field) + + except Exception as e: + # Mark field for full_context retry and preserve error info + field.field_type = "full_context" + logging.error( + f"Error on {field.field_name}: {type(e).__name__}: {str(e)}, marking as full_context" + ) + return (field.field_name, "N/A", field) + diff --git a/src/pipelines/saas/prompts/prompt_calls.py b/src/pipelines/saas/prompts/prompt_calls.py index 6724a97..c3032f0 100644 --- a/src/pipelines/saas/prompts/prompt_calls.py +++ b/src/pipelines/saas/prompts/prompt_calls.py @@ -4,7 +4,8 @@ import src.config as config import src.prompts.prompt_templates as prompt_templates from src.constants.constants import Constants from src.prompts.fieldset import Field, FieldSet -from src.utils import llm_utils, string_utils +from src.utils import llm_utils, string_utils, timing_utils, rag_utils +from src.pipelines.shared.preprocessing.hybrid_smart_chunking_funcs import format_chunks_for_llm import json @@ -865,3 +866,93 @@ def provider_name_match_check( f"[is_provider_name_match_llm] Error calling LLM: {e}, defaulting to False" ) return False + + +def prompt_hsc_single_field( + field, retriever, reranker, rag_config, text_dict, constants, filename +): + """Process a single field in parallel for HSC.""" + try: + retrieval_query = field.retrieval_question + + if not retrieval_query: + return (field.field_name, "N/A", field) + + # Retrieve + rerank using retrieval question + with timing_utils.timed_block( + f"hsc.field.{field.field_name}.retrieval", context=filename + ): + retrieved = retriever.invoke(retrieval_query)[: rag_config.ENSEMBLE_TOP_K] + reranked = rag_utils.rerank_documents( + retrieval_query, + retrieved, + reranker, + rag_config.RERANKER_TOP_K, + rag_config.RERANKER, + ) + + # Build context and use actual LLM prompt for extraction + context = format_chunks_for_llm(text_dict, reranked) + + if len(context) == 0: + # Fallback: mark field for full_context processing + field.field_type = "full_context" + logging.warning( + f"No context retrieved for {field.field_name}, marking as full_context" + ) + return (field.field_name, "N/A", field) + + llm_prompt = field.get_prompt_dict( + constants + ) # Returns dict of {field_name : prompt} + prompt, _parser = prompt_templates.ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(context, llm_prompt) + + logging.debug(f"Field: {field.field_name}") + logging.debug(f"Retrieval question: {retrieval_query}") + logging.debug(f"LLM prompt: {llm_prompt}") + + # LLM call + with timing_utils.timed_block( + f"hsc.field.{field.field_name}.llm_call", context=filename + ): + llm_answer_raw = llm_utils.invoke_claude( + prompt, "sonnet_latest", filename, max_tokens=8192 + ) + try: + llm_answer_final = _parser(llm_answer_raw) # returns dict + + val = llm_answer_final.get(field.field_name) + if isinstance(val, list): + llm_answer_final[field.field_name] = val + elif isinstance(val, str): + llm_answer_final[field.field_name] = [val] + + field_value = llm_answer_final[field.field_name] + if field_value[0] == "N/A": + field.field_type = "full_context" + if field.field_name == "PAYER_NAME": + field.prompt = ( + field.prompt + + prompt_templates.FULL_CONTEXT_PAYER_NAME_ADDITIONAL_INSTRUCTION() + ) + logging.warning( + f"Obtained N/A for {field.field_name}, marking as full_context" + ) + + return (field.field_name, field_value, field) + + except Exception as e: + logging.warning( + f"Failed to parse JSON response for field {field.field_name}: " + f"{type(e).__name__}: {str(e)}. Setting field value to N/A." + ) + return (field.field_name, "N/A", field) + + except Exception as e: + # Mark field for full_context retry and preserve error info + field.field_type = "full_context" + logging.error( + f"Error on {field.field_name}: {type(e).__name__}: {str(e)}, marking as full_context" + ) + return (field.field_name, "N/A", field) + diff --git a/src/pipelines/shared/preprocessing/hybrid_smart_chunking_funcs.py b/src/pipelines/shared/preprocessing/hybrid_smart_chunking_funcs.py index cc634cb..b2bff65 100644 --- a/src/pipelines/shared/preprocessing/hybrid_smart_chunking_funcs.py +++ b/src/pipelines/shared/preprocessing/hybrid_smart_chunking_funcs.py @@ -28,103 +28,13 @@ from src.pipelines.shared.extraction.one_to_one_funcs import ( from src.utils.rag_utils import ( RAGConfig, HSC_CONFIG, - get_bedrock_client, get_reranker_client, get_embeddings, create_vectorstore, create_hybrid_retriever, - rerank_documents, ) -def process_single_field( - field, retriever, reranker, rag_config, text_dict, constants, filename -): - """Process a single field in parallel for HSC.""" - try: - retrieval_query = field.retrieval_question - - if not retrieval_query: - return (field.field_name, "N/A", field) - - # Retrieve + rerank using retrieval question - with timing_utils.timed_block( - f"hsc.field.{field.field_name}.retrieval", context=filename - ): - retrieved = retriever.invoke(retrieval_query)[: rag_config.ENSEMBLE_TOP_K] - reranked = rerank_documents( - retrieval_query, - retrieved, - reranker, - rag_config.RERANKER_TOP_K, - rag_config.RERANKER, - ) - - # Build context and use actual LLM prompt for extraction - context = format_chunks_for_llm(text_dict, reranked) - - if len(context) == 0: - # Fallback: mark field for full_context processing - field.field_type = "full_context" - logging.warning( - f"No context retrieved for {field.field_name}, marking as full_context" - ) - return (field.field_name, "N/A", field) - - llm_prompt = field.get_prompt_dict( - constants - ) # Returns dict of {field_name : prompt} - prompt = prompt_templates.ONE_TO_ONE_SINGLE_FIELD_TEMPLATE(context, llm_prompt) - - logging.debug(f"Field: {field.field_name}") - logging.debug(f"Retrieval question: {retrieval_query}") - logging.debug(f"LLM prompt: {llm_prompt}") - - # LLM call - with timing_utils.timed_block( - f"hsc.field.{field.field_name}.llm_call", context=filename - ): - raw = llm_utils.invoke_claude( - prompt, "sonnet_latest", filename, max_tokens=8192 - ) - try: - parsed = string_utils.universal_json_load(raw) # returns dict - - val = parsed.get(field.field_name) - if isinstance(val, list): - parsed[field.field_name] = "|".join(map(str, val)) - elif isinstance(val, str): - parsed[field.field_name] = val - field_value = parsed[field.field_name] - if field_value == "N/A": - field.field_type = "full_context" - if field.field_name == "PAYER_NAME": - field.prompt = ( - field.prompt - + prompt_templates.FULL_CONTEXT_PAYER_NAME_ADDITIONAL_INSTRUCTION() - ) - logging.warning( - f"Obtained N/A for {field.field_name}, marking as full_context" - ) - - return (field.field_name, field_value, field) - - except Exception as e: - logging.warning( - f"Failed to parse JSON response for field {field.field_name}: " - f"{type(e).__name__}: {str(e)}. Setting field value to N/A." - ) - return (field.field_name, "N/A", field) - - except Exception as e: - # Mark field for full_context retry and preserve error info - field.field_type = "full_context" - logging.error( - f"Error on {field.field_name}: {type(e).__name__}: {str(e)}, marking as full_context" - ) - return (field.field_name, "N/A", field) - - def clip_context_by_tokens(context: str, max_tokens: int = 150000) -> str: """ Clip context text if it exceeds max_tokens. @@ -315,7 +225,7 @@ def run_hybrid_smart_chunked_fields( ) as executor: futures = [ executor.submit( - process_single_field, + prompt_hsc_single_field, field, retriever, reranker,