jacob-mathison/doczyai-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Resolve circular import of prompt_hsc_single_field

Browse Source
This commit is contained in:
Katon Minhas
2026-02-02 21:57:22 -05:00
parent bb440acd7e
commit 9b30c5f282
4 changed files with 91 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pipelines/clients/bcbs_promise/prompts/prompt_calls.py
-89
View File
@@ -852,92 +852,3 @@ 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)
src/pipelines/clients/clover/prompts/prompt_calls.py
+1 -90
View File
@@ -906,93 +906,4 @@ def provider_name_match_check(
logging.error(
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)
return False
src/pipelines/saas/prompts/prompt_calls.py
-89
View File
@@ -867,92 +867,3 @@ 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)
src/pipelines/shared/preprocessing/hybrid_smart_chunking_funcs.py
+90 -4
View File
@@ -14,12 +14,9 @@ from typing import List, Tuple, Dict
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
import src.utils.llm_utils as llm_utils
import src.utils.string_utils as string_utils
import src.utils.timing_utils as timing_utils
from src.utils import llm_utils, string_utils, timing_utils, rag_utils
import src.pipelines.shared.preprocessing.preprocessing_funcs as preprocessing_funcs
import src.pipelines.shared.preprocessing.hybrid_smart_chunking_preprocessing as hybrid_smart_chunking_preprocessing
from src.pipelines.saas.prompts.prompt_calls import prompt_hsc_single_field
from src import config
from src.prompts import prompt_templates
from src.prompts.fieldset import FieldSet
@@ -338,3 +335,92 @@ def extract_amendment_num_from_filename(answer_dict: dict, filename: str) -> dic
answer_dict["FILENAME_AMENDMENT_NUM"] = "N/A"
return answer_dict
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)