diff --git a/.gitignore b/.gitignore index 2bade91..4a8a85f 100644 --- a/.gitignore +++ b/.gitignore @@ -45,6 +45,7 @@ Thumbs.db !src/constants/**/*.csv !src/constants/**/*.json !src/prompts/*.json +!src/qc_qa/confidence/*.json !requirements.txt # Build artifacts diff --git a/src/config.py b/src/config.py index 687a518..f3d5c7b 100644 --- a/src/config.py +++ b/src/config.py @@ -211,6 +211,11 @@ AC_DF = get_arg_value("ac_df", None) B_DF = get_arg_value("b_df", None) DF_READ_MODE = get_arg_value("df_read_mode", "s3") +# Confidence scoring (1:1) reporting threshold. Per-field _CONF values +# below this land in CONFIDENCE-FLAGGED.csv and count toward the +# pct_below_threshold column in CONFIDENCE-SUMMARY.csv. +CONFIDENCE_THRESHOLD = float(get_arg_value("confidence_threshold", "0.6")) + ######################################## FILE LOG ######################################## diff --git a/src/pipelines/runner.py b/src/pipelines/runner.py index 6feb40f..73bf9b5 100644 --- a/src/pipelines/runner.py +++ b/src/pipelines/runner.py @@ -452,6 +452,20 @@ def main(client: str = "saas", testing=False, test_params={}): config.BATCH_ID, run_timestamp ) + # ========== Confidence reporting (DAIP2-2695) ========== + # Build per-field distribution + flagged-for-review DataFrames from + # any _CONF columns that landed in FINAL_RESULT_DF_CC. Both + # frames may be empty when no scoring took place (e.g. no 1:1 fields + # in the field set); in that case we skip writing. + from src.qc_qa.confidence.summary import compute_summary, compute_flagged + + confidence_summary_df = compute_summary( + FINAL_RESULT_DF_CC, threshold=config.CONFIDENCE_THRESHOLD + ) + confidence_flagged_df = compute_flagged( + FINAL_RESULT_DF_CC, threshold=config.CONFIDENCE_THRESHOLD + ) + with timing_utils.timed_block("write_outputs", log_level="INFO"): if config.WRITE_TO_S3: doczy_output_for_pc = io_utils.write_s3( @@ -480,6 +494,21 @@ def main(client: str = "saas", testing=False, test_params={}): if qc_qa_stats_df is not None: io_utils.write_s3(qc_qa_stats_df, "", run_timestamp, "qc_qa_stats") + if not confidence_summary_df.empty: + io_utils.write_s3( + confidence_summary_df, + "", + run_timestamp, + "confidence_summary", + ) + if not confidence_flagged_df.empty: + io_utils.write_s3( + confidence_flagged_df, + "", + run_timestamp, + "confidence_flagged", + ) + if not per_file_usage_df.empty and not batch_summary_df.empty: logging.info("Exporting usage and cost tracking data to S3...") io_utils.write_s3(per_file_usage_df, "", run_timestamp, "usage") @@ -518,6 +547,21 @@ def main(client: str = "saas", testing=False, test_params={}): qc_qa_stats_df, "", run_timestamp, "qc_qa_stats" ) + if not confidence_summary_df.empty: + io_utils.write_local( + confidence_summary_df, + "", + run_timestamp, + "confidence_summary", + ) + if not confidence_flagged_df.empty: + io_utils.write_local( + confidence_flagged_df, + "", + run_timestamp, + "confidence_flagged", + ) + if config.PERFORM_PARENT_CHILD_MAPPING: if FINAL_RESULT_DF_CC.empty: logging.warning( diff --git a/src/pipelines/saas/file_processing.py b/src/pipelines/saas/file_processing.py index b1de540..e140ea9 100644 --- a/src/pipelines/saas/file_processing.py +++ b/src/pipelines/saas/file_processing.py @@ -278,15 +278,22 @@ def run_one_to_one_prompts( ################## RUN HYBRID SMART CHUNKED PROMPTS ################## # RAG function loads retrieval questions internally and matches with investment_prompts.json # Only run if there are fields to extract - hybrid_smart_chunked_answers_dict = {} + hybrid_smart_chunked_answers_dict: dict = {} + hybrid_smart_chunked_metadata_dict: 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 - ) + ( + hybrid_smart_chunked_answers_dict, + hybrid_smart_chunked_metadata_dict, + ) = hybrid_smart_chunking_funcs.run_hybrid_smart_chunked_fields( + one_to_one_fields, + constants, + contract_text, + filename, + text_dict, + return_metadata=True, ) logging.debug( f"Hybrid Smart Chunked Answers for {filename}: {hybrid_smart_chunked_answers_dict}" @@ -300,6 +307,32 @@ def run_one_to_one_prompts( # Add HSC's N/A if field doesn't exist yet one_to_one_results[key] = value + ################## SCORE PER-FIELD CONFIDENCE ################## + # T2 (DAIP2-2692): rule-based confidence per HSC field. Combines + # exact / token-overlap / fuzzy / number-match / regex / grounding / + # na-check signals with field-type-aware weights. T3 will layer an + # LLM verifier on top for gray-zone scores. + from src.qc_qa.confidence import score_field + + for field_name, metadata in hybrid_smart_chunked_metadata_dict.items(): + meta = metadata if isinstance(metadata, dict) else {} + # Pull the field's prompt for the na_check signal's keyword set. + field_obj = ( + one_to_one_fields.get_field(field_name) + if one_to_one_fields.contains_fields() + else None + ) + field_prompt = getattr(field_obj, "prompt", "") if field_obj else "" + + rule_score = score_field( + field_name=field_name, + value=one_to_one_results.get(field_name), + metadata=meta, + contract_text=contract_text, + field_prompt=field_prompt, + ) + one_to_one_results[f"{field_name}_CONF"] = rule_score + ################## RUN CLIENT FULL-CONTEXT FIELDS ################## # Client-specific full_context fields (e.g. Clover's 12 audit/filing/CA fields) if client_full_context_fields is not None: diff --git a/src/pipelines/shared/postprocessing/postprocess.py b/src/pipelines/shared/postprocessing/postprocess.py index 6caf9cd..2a7b6aa 100644 --- a/src/pipelines/shared/postprocessing/postprocess.py +++ b/src/pipelines/shared/postprocessing/postprocess.py @@ -90,6 +90,12 @@ def standard_postprocess(df, constants: Constants): ) for col in df.columns: + # _CONF columns hold per-field confidence floats (DAIP2-2692). They + # must not be touched by any of the value-shape normalizers below, + # which substring-match column names and would otherwise corrupt + # e.g. AUTO_RENEWAL_IND_CONF (matches "_IND") into a Y/N string. + if col.endswith("_CONF"): + continue if "_IND" in col: df[col] = df[col].apply(postprocessing_funcs.normalize_indicator_field) # Normalize _IND fields and clean up TIN/NPI fields diff --git a/src/pipelines/shared/postprocessing/postprocessing_funcs.py b/src/pipelines/shared/postprocessing/postprocessing_funcs.py index 2f9d642..e144115 100644 --- a/src/pipelines/shared/postprocessing/postprocessing_funcs.py +++ b/src/pipelines/shared/postprocessing/postprocessing_funcs.py @@ -483,7 +483,11 @@ def reorder_columns( Steps: 1. Adds any missing columns from `column_order` to the DataFrame at once, filled with empty strings. 2. Reorders the columns of the DataFrame to match `column_order`. - 3. Drops any columns in the DataFrame that are not in `column_order`. + 3. Drops any columns in the DataFrame that are not in `column_order`, + except columns ending with `_CONF` which are preserved at the end. + This carve-out lets the one-to-one confidence-scoring stage emit + per-field _CONF columns without registering each one in + FIELD_FORMAT_MAPPING. Args: df (pd.DataFrame): The input DataFrame. @@ -510,8 +514,15 @@ def reorder_columns( [df_copy, pd.DataFrame(empty_cols, index=df_copy.index)], axis=1 ) - # Keep only columns in column_order; drop any extras not in the mapping + # Keep only columns in column_order; drop any extras not in the mapping, + # except *_CONF columns which are preserved (sorted) at the end. final_column_order = [col for col in column_order if col in df_copy.columns] + conf_columns = sorted( + col + for col in df_copy.columns + if col.endswith("_CONF") and col not in column_order + ) + final_column_order.extend(conf_columns) # Return the DataFrame with reordered columns return df_copy[final_column_order] diff --git a/src/pipelines/shared/preprocessing/hybrid_smart_chunking_funcs.py b/src/pipelines/shared/preprocessing/hybrid_smart_chunking_funcs.py index f695405..aaba82c 100644 --- a/src/pipelines/shared/preprocessing/hybrid_smart_chunking_funcs.py +++ b/src/pipelines/shared/preprocessing/hybrid_smart_chunking_funcs.py @@ -269,11 +269,18 @@ def run_hybrid_smart_chunked_fields( filename: str, text_dict: dict, rag_config: RAGConfig = HSC_CONFIG, -) -> dict[str, str]: + return_metadata: bool = False, +): """Process fields using RAG: hybrid retrieval (BM25+semantic) + reranking. Uses retrieval questions from the field's retrieval_question attribute for chunk retrieval, then actual LLM prompts from the field's prompt attribute for field extraction. + + By default returns a dict of {field_name: value} for backwards compatibility. + When return_metadata=True, returns a tuple (answers_dict, metadata_dict) where + metadata_dict maps each field_name to {confidence, verdict, supporting_snippet, + retrieved_chunk_count, retrieved_chunk_ids}. Used by the one-to-one confidence + scoring stage downstream. """ logging.debug(f"Starting RAG extraction for {filename}") @@ -291,7 +298,7 @@ def run_hybrid_smart_chunked_fields( logging.debug( f"No fields with retrieval questions found | Filename: {filename}" ) - return {} + return ({}, {}) if return_metadata else {} try: # Setup: clients, splitter, chunks @@ -316,7 +323,7 @@ def run_hybrid_smart_chunked_fields( gc.collect() if not chunks: - return {} + return ({}, {}) if return_metadata else {} # Build retriever with timing_utils.timed_block("hsc.vectorstore_creation", context=filename): @@ -324,6 +331,7 @@ def run_hybrid_smart_chunked_fields( retriever = create_hybrid_retriever(chunks, vectorstore, rag_config) answers_dict = {} + metadata_dict: dict[str, dict] = {} with timing_utils.timed_block( "hsc.parallel_field_processing", context=filename ): @@ -347,8 +355,9 @@ def run_hybrid_smart_chunked_fields( for future in concurrent.futures.as_completed(futures): try: - field_name, field_value, field = future.result() + field_name, field_value, field, metadata = future.result() answers_dict[field_name] = field_value + metadata_dict[field_name] = metadata except Exception as e: logging.error(f"Error processing field: {str(e)}") @@ -381,11 +390,11 @@ def run_hybrid_smart_chunked_fields( answers_dict, field_name="PAYER_STATE" ) - return answers_dict + return (answers_dict, metadata_dict) if return_metadata else answers_dict except Exception as e: logging.error(f"RAG pipeline failed: {e}") - return {} + return ({}, {}) if return_metadata else {} def extract_amendment_num_from_filename(answer_dict: dict, filename: str) -> dict: @@ -427,15 +436,96 @@ def extract_amendment_num_from_filename(answer_dict: dict, filename: str) -> dic return answer_dict +# Cap on the joined retrieved-chunks blob we keep in metadata. The grounding +# signal scans this text for the extracted value; 12k chars is enough to cover +# the typical ~10 reranked chunks without ballooning memory or making the +# downstream search slow. +_RETRIEVED_CHUNKS_TEXT_CAP = 12_000 + + +def _empty_hsc_metadata() -> dict: + """Default metadata payload for HSC fields that bail out before extracting.""" + return { + "confidence": None, + "verdict": "not_found", + "supporting_snippet": "", + "retrieved_chunk_count": 0, + "retrieved_chunk_ids": [], + "retrieved_chunks_text": "", + } + + +def _extract_hsc_metadata(parsed: dict, reranked) -> dict: + """Pull confidence / verdict / supporting_snippet out of the parsed LLM dict + and attach a summary of the retrieved chunks the LLM saw. + + The chunks summary includes both lightweight ids (for traceability) and + the joined chunk text (for the grounding signal in the rule-based scorer + downstream). Text is capped so the metadata stays bounded. + + Stays defensive: any of the keys may be missing or the wrong type if the LLM + didn't follow the prompt exactly. We coerce/default rather than fail. + """ + raw_conf = parsed.get("confidence") if isinstance(parsed, dict) else None + try: + conf = float(raw_conf) if raw_conf is not None else None + if conf is not None: + conf = max(0.0, min(1.0, conf)) + except (TypeError, ValueError): + conf = None + + verdict = parsed.get("verdict") if isinstance(parsed, dict) else None + if verdict not in {"correct", "uncertain", "not_found"}: + verdict = "uncertain" if conf is not None else "not_found" + + snippet = parsed.get("supporting_snippet") if isinstance(parsed, dict) else None + snippet = str(snippet) if snippet is not None else "" + + chunk_ids: list[str] = [] + chunk_texts: list[str] = [] + if reranked: + for doc in reranked: + metadata = getattr(doc, "metadata", {}) or {} + chunk_id = ( + metadata.get("chunk_id") + or metadata.get("id") + or metadata.get("source") + or "" + ) + if chunk_id: + chunk_ids.append(str(chunk_id)) + text = getattr(doc, "page_content", "") or "" + if text: + chunk_texts.append(str(text)) + + joined_text = "\n\n".join(chunk_texts) + if len(joined_text) > _RETRIEVED_CHUNKS_TEXT_CAP: + joined_text = joined_text[:_RETRIEVED_CHUNKS_TEXT_CAP] + + return { + "confidence": conf, + "verdict": verdict, + "supporting_snippet": snippet[:500], # cap so it doesn't bloat the CSV + "retrieved_chunk_count": len(reranked) if reranked else 0, + "retrieved_chunk_ids": chunk_ids, + "retrieved_chunks_text": joined_text, + } + + def prompt_hsc_single_field( field, retriever, reranker, rag_config, text_dict, constants, filename ): - """Process a single field in parallel for HSC.""" + """Process a single field in parallel for HSC. + + Returns a 4-tuple: (field_name, field_value, field, metadata) where + metadata is the dict produced by _extract_hsc_metadata() — used by the + one-to-one confidence scoring stage downstream. + """ try: retrieval_query = field.retrieval_question if not retrieval_query: - return (field.field_name, "N/A", field) + return (field.field_name, "N/A", field, _empty_hsc_metadata()) # Retrieve + rerank using retrieval question with timing_utils.timed_block( @@ -457,7 +547,7 @@ def prompt_hsc_single_field( logging.warning( f"No context retrieved for {field.field_name}, marking as Null value" ) - return (field.field_name, "N/A", field) + return (field.field_name, "N/A", field, _empty_hsc_metadata()) llm_prompt = field.get_prompt_dict( constants @@ -501,15 +591,16 @@ def prompt_hsc_single_field( try: llm_answer_final = _parser(llm_answer_raw) # returns dict field_value = llm_answer_final.get(field.field_name) - return (field.field_name, field_value, field) + metadata = _extract_hsc_metadata(llm_answer_final, reranked) + return (field.field_name, field_value, field, metadata) 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) + return (field.field_name, "N/A", field, _empty_hsc_metadata()) except Exception as e: logging.error(f"Error on {field.field_name}: {type(e).__name__}: {str(e)}") - return (field.field_name, "N/A", field) + return (field.field_name, "N/A", field, _empty_hsc_metadata()) diff --git a/src/prompts/prompt_templates.py b/src/prompts/prompt_templates.py index 0c28ae6..6d1d708 100644 --- a/src/prompts/prompt_templates.py +++ b/src/prompts/prompt_templates.py @@ -2610,7 +2610,13 @@ Answer the following question: {field_prompt} {context} ## END CONTRACT TEXT ## -Briefly explain your answer, then put the final answer in a properly-formatted JSON dictionary, where the key is the field name and value is the answer. If there is no valid answer, return "N/A". +Briefly explain your answer, then put the final answer in a properly-formatted JSON dictionary with these keys: +- The field name (key) mapped to the extracted value (or "N/A" if not found). +- "confidence": a number between 0.0 and 1.0 expressing how confident you are in the extracted value (1.0 = certain, 0.0 = guess). +- "verdict": one of "correct" / "uncertain" / "not_found". +- "supporting_snippet": the shortest exact text from the contract that supports your answer (max ~200 chars), or "" if the answer is "N/A". + +If there is no valid answer, return "N/A" as the field value, set "verdict" to "not_found", and "supporting_snippet" to "". """ # Use field-aware parser if field_name is provided diff --git a/src/qc_qa/confidence/__init__.py b/src/qc_qa/confidence/__init__.py new file mode 100644 index 0000000..65a668c --- /dev/null +++ b/src/qc_qa/confidence/__init__.py @@ -0,0 +1,16 @@ +"""One-to-one field confidence scoring. + +Stages (per the implementation plan): + 1. signals - rule-based per-field signals (this ticket: DAIP2-2692) + 2. llm_verifier - gray-zone LLM check (DAIP2-2693) + 3. calibration - LLM-stated -> empirical accuracy mapping (DAIP2-2694) + 4. report - field-level distribution stats (DAIP2-2695) + +This module is intentionally self-contained. It consumes the metadata dict +produced by `prompt_hsc_single_field` (see hybrid_smart_chunking_funcs.py) +and emits a per-field rule score in [0, 1]. +""" + +from src.qc_qa.confidence.scorer import score_field + +__all__ = ["score_field"] diff --git a/src/qc_qa/confidence/calibration.py b/src/qc_qa/confidence/calibration.py new file mode 100644 index 0000000..7e2ac54 --- /dev/null +++ b/src/qc_qa/confidence/calibration.py @@ -0,0 +1,161 @@ +"""Confidence calibration: map raw scores -> empirically-calibrated scores. + +LLM-stated and even rule-based confidence scores tend to be miscalibrated: +a "0.9" rule score does not always correspond to 90% empirical accuracy. +This module applies a per-field-type calibration curve so a final score +of 0.X reflects roughly an X% probability of being correct in production. + +The calibration curve is a small JSON table of (raw, calibrated) anchor +points per field type. Lookup is linear interpolation. Out-of-domain +inputs (NaN, negative, > 1.0) are clamped before lookup. + +Defaults ship as identity (raw == calibrated), so this stage is a no-op +until a real curve is backfilled from QC data. That keeps the pipeline +safe to ship before we have enough labeled samples to calibrate against. + +To regenerate the table when QC data is available, run: + uv run python -m src.qc_qa.confidence.generate_calibration_table +(That script will land alongside this module when the data pipeline is +ready; until then the identity table is correct behavior.) +""" + +from __future__ import annotations + +import bisect +import json +import logging +import os +from typing import Dict, List, Optional, Tuple + +from src.qc_qa.confidence.field_types import FieldType + +logger = logging.getLogger(__name__) + +# Anchor point: (raw_score, calibrated_score) +_AnchorPoint = Tuple[float, float] + +_TABLE_FILENAME = "calibration_table.json" +_TABLE_PATH = os.path.join(os.path.dirname(__file__), _TABLE_FILENAME) + +# Module-level cache of (field_type_key -> sorted list of (raw, calibrated)) +_CACHED_TABLE: Optional[Dict[str, List[_AnchorPoint]]] = None + +# Identity fallback used if the file is missing or unreadable. Never raises; +# a broken file should not break extraction. +_IDENTITY: List[_AnchorPoint] = [(0.0, 0.0), (1.0, 1.0)] + + +def _parse_table(raw: dict) -> Dict[str, List[_AnchorPoint]]: + """Validate + normalize the JSON shape into sorted anchor lists. + + Anything malformed (wrong types, fewer than two points, points outside + [0, 1]) silently falls back to identity for that key. Calibration must + never make scoring worse than no calibration. + """ + parsed: Dict[str, List[_AnchorPoint]] = {} + for key, points in raw.items(): + if key.startswith("_"): + continue # metadata fields like _comment, _schema_version + if not isinstance(points, list) or len(points) < 2: + parsed[key] = list(_IDENTITY) + continue + cleaned: List[_AnchorPoint] = [] + for pt in points: + if not isinstance(pt, (list, tuple)) or len(pt) != 2: + continue + try: + raw_v = float(pt[0]) + cal_v = float(pt[1]) + except (TypeError, ValueError): + continue + if 0.0 <= raw_v <= 1.0 and 0.0 <= cal_v <= 1.0: + cleaned.append((raw_v, cal_v)) + if len(cleaned) < 2: + parsed[key] = list(_IDENTITY) + continue + cleaned.sort(key=lambda p: p[0]) + # Ensure the curve covers the full [0, 1] domain. If the lowest + # anchor is > 0 or the highest is < 1, extend with identity ends so + # the interpolation never extrapolates. + if cleaned[0][0] > 0.0: + cleaned.insert(0, (0.0, cleaned[0][1])) + if cleaned[-1][0] < 1.0: + cleaned.append((1.0, cleaned[-1][1])) + parsed[key] = cleaned + return parsed + + +def _load_table() -> Dict[str, List[_AnchorPoint]]: + """Load the calibration table from disk. Caches the result; safe to + call repeatedly. Falls back to identity-for-default on any failure. + """ + global _CACHED_TABLE + if _CACHED_TABLE is not None: + return _CACHED_TABLE + + try: + with open(_TABLE_PATH, "r", encoding="utf-8") as f: + raw = json.load(f) + _CACHED_TABLE = _parse_table(raw) + if "default" not in _CACHED_TABLE: + _CACHED_TABLE["default"] = list(_IDENTITY) + except FileNotFoundError: + logger.info( + "Calibration table not found at %s; using identity. " + "This is expected before QC backfill.", + _TABLE_PATH, + ) + _CACHED_TABLE = {"default": list(_IDENTITY)} + except Exception as exc: + logger.warning( + "Failed to load calibration table from %s (%s); using identity.", + _TABLE_PATH, + exc, + ) + _CACHED_TABLE = {"default": list(_IDENTITY)} + return _CACHED_TABLE + + +def reload_table() -> None: + """Clear the cached table so the next calibrate() call re-reads the JSON. + Intended for tests and offline regeneration workflows. + """ + global _CACHED_TABLE + _CACHED_TABLE = None + + +def _interpolate(points: List[_AnchorPoint], x: float) -> float: + """Linear interpolation between sorted (raw, calibrated) anchors.""" + raws = [p[0] for p in points] + idx = bisect.bisect_left(raws, x) + if idx == 0: + return points[0][1] + if idx >= len(points): + return points[-1][1] + x0, y0 = points[idx - 1] + x1, y1 = points[idx] + if x1 == x0: + return y0 + return y0 + (y1 - y0) * (x - x0) / (x1 - x0) + + +def calibrate(field_type: FieldType | str, raw_score: float) -> float: + """Map a raw confidence in [0, 1] to a calibrated confidence in [0, 1]. + + Always returns a finite float in [0, 1]. Bad inputs return the raw + value clamped (i.e. the calibration is a no-op on the failure path). + """ + # Coerce + clamp input. + try: + x = float(raw_score) + except (TypeError, ValueError): + return 0.0 + if x != x: # NaN + return 0.0 + x = max(0.0, min(1.0, x)) + + table = _load_table() + key = field_type.value if isinstance(field_type, FieldType) else str(field_type) + points = table.get(key) or table.get("default") or _IDENTITY + out = _interpolate(points, x) + return max(0.0, min(1.0, out)) diff --git a/src/qc_qa/confidence/calibration_table.json b/src/qc_qa/confidence/calibration_table.json new file mode 100644 index 0000000..5eb9bd3 --- /dev/null +++ b/src/qc_qa/confidence/calibration_table.json @@ -0,0 +1,15 @@ +{ + "_comment": "Per-field-type calibration curves: (raw_score, calibrated_score) anchor points. Linearly interpolated between adjacent points; values outside [0,1] are clamped. The defaults ship as identity (raw == calibrated) so the pipeline is a no-op until a real curve is backfilled from QC data. Regenerate with src/qc_qa/confidence/generate_calibration_table.py once enough labeled samples are available.", + "_schema_version": 1, + "_generated_at": null, + "_generated_from": null, + "default": [[0.0, 0.0], [1.0, 1.0]], + "date": [[0.0, 0.0], [1.0, 1.0]], + "money": [[0.0, 0.0], [1.0, 1.0]], + "tin": [[0.0, 0.0], [1.0, 1.0]], + "npi": [[0.0, 0.0], [1.0, 1.0]], + "code": [[0.0, 0.0], [1.0, 1.0]], + "name": [[0.0, 0.0], [1.0, 1.0]], + "boolean": [[0.0, 0.0], [1.0, 1.0]], + "free_text":[[0.0, 0.0], [1.0, 1.0]] +} diff --git a/src/qc_qa/confidence/field_types.py b/src/qc_qa/confidence/field_types.py new file mode 100644 index 0000000..f72c027 --- /dev/null +++ b/src/qc_qa/confidence/field_types.py @@ -0,0 +1,145 @@ +"""Per-field-type registry and signal weight presets. + +Each one-to-one HSC field maps to a FieldType. Each FieldType has a preset +of signal weights tuned for that kind of value: + + - DATE -> regex + number_match dominate + - MONEY -> number_match dominates + - TIN/NPI -> exact + regex dominate + - CODE -> exact dominates + - NAME -> fuzzy + token_overlap dominate + - BOOLEAN -> exact + na_check dominate + - FREE_TEXT -> token_overlap + fuzzy + grounding (default) + +Signals not listed for a type get weight 0 and are skipped at scoring time. +The scorer renormalizes over the signals it actually computed, so a missing +signal (e.g. no retrieved chunks for grounding) does not bias the result. +""" + +from __future__ import annotations + +from enum import Enum +from typing import Dict + + +class FieldType(str, Enum): + DATE = "date" + MONEY = "money" + TIN = "tin" + NPI = "npi" + CODE = "code" + NAME = "name" + BOOLEAN = "boolean" + FREE_TEXT = "free_text" + + +# Map field name -> FieldType. Defaults to FREE_TEXT for anything not listed. +# Keep this list narrow on purpose; new fields added here should have an +# obvious, single-type interpretation. When in doubt, leave it as FREE_TEXT +# and let the default weights apply. +FIELD_TYPE_MAP: Dict[str, FieldType] = { + # Dates + "EFFECTIVE_DT": FieldType.DATE, + "TERMINATION_DT": FieldType.DATE, + "AARETE_DERIVED_EFFECTIVE_DT": FieldType.DATE, + "AARETE_DERIVED_TERMINATION_DT": FieldType.DATE, + "REIMB_EFFECTIVE_DT": FieldType.DATE, + "REIMB_TERMINATION_DT": FieldType.DATE, + # Names + "PAYER_NAME": FieldType.NAME, + "AARETE_DERIVED_PAYER_NAME": FieldType.NAME, + "PROVIDER_NAME": FieldType.NAME, + "AARETE_DERIVED_PROVIDER_NAME": FieldType.NAME, + "CONTRACT_TITLE": FieldType.NAME, + # Boolean / indicator + "AUTO_RENEWAL_IND": FieldType.BOOLEAN, + # Free text (default category, listed for clarity) + "AUTO_RENEWAL_TERM": FieldType.FREE_TEXT, + "PROVIDER_STATE": FieldType.FREE_TEXT, + "PAYER_STATE": FieldType.FREE_TEXT, + "CONTRACT_AMENDMENT_NUM": FieldType.FREE_TEXT, +} + + +# Signal names used as keys throughout the confidence package. +EXACT = "exact" +TOKEN_OVERLAP = "token_overlap" +FUZZY = "fuzzy" +NUMBER_MATCH = "number_match" +REGEX = "regex" +GROUNDING = "grounding" +NA_CHECK = "na_check" + +ALL_SIGNALS = (EXACT, TOKEN_OVERLAP, FUZZY, NUMBER_MATCH, REGEX, GROUNDING, NA_CHECK) + + +# Per-FieldType weight presets. Weights are relative; the scorer renormalizes +# to whatever signals actually fire for a given (value, context) pair. +WEIGHT_PRESETS: Dict[FieldType, Dict[str, float]] = { + FieldType.DATE: { + REGEX: 0.30, + NUMBER_MATCH: 0.25, + GROUNDING: 0.25, + EXACT: 0.05, + FUZZY: 0.05, + NA_CHECK: 0.10, + }, + FieldType.MONEY: { + NUMBER_MATCH: 0.35, + GROUNDING: 0.25, + REGEX: 0.20, + TOKEN_OVERLAP: 0.05, + EXACT: 0.05, + NA_CHECK: 0.10, + }, + FieldType.TIN: { + EXACT: 0.35, + REGEX: 0.30, + NUMBER_MATCH: 0.20, + GROUNDING: 0.10, + NA_CHECK: 0.05, + }, + FieldType.NPI: { + EXACT: 0.35, + REGEX: 0.30, + NUMBER_MATCH: 0.20, + GROUNDING: 0.10, + NA_CHECK: 0.05, + }, + FieldType.CODE: { + EXACT: 0.40, + REGEX: 0.20, + GROUNDING: 0.20, + TOKEN_OVERLAP: 0.10, + NA_CHECK: 0.10, + }, + FieldType.NAME: { + FUZZY: 0.30, + GROUNDING: 0.30, + TOKEN_OVERLAP: 0.20, + EXACT: 0.10, + NA_CHECK: 0.10, + }, + FieldType.BOOLEAN: { + EXACT: 0.40, + GROUNDING: 0.30, + NA_CHECK: 0.30, + }, + FieldType.FREE_TEXT: { + TOKEN_OVERLAP: 0.30, + FUZZY: 0.25, + GROUNDING: 0.25, + EXACT: 0.10, + NA_CHECK: 0.10, + }, +} + + +def get_field_type(field_name: str) -> FieldType: + """Resolve a field name to its FieldType. Defaults to FREE_TEXT.""" + return FIELD_TYPE_MAP.get(field_name, FieldType.FREE_TEXT) + + +def get_weight_preset(field_type: FieldType) -> Dict[str, float]: + """Return the (signal -> weight) preset for a field type.""" + return WEIGHT_PRESETS[field_type] diff --git a/src/qc_qa/confidence/scorer.py b/src/qc_qa/confidence/scorer.py new file mode 100644 index 0000000..6d07ad6 --- /dev/null +++ b/src/qc_qa/confidence/scorer.py @@ -0,0 +1,299 @@ +"""Combine rule-based signals into a per-field confidence score. + +Inputs: + field_name : the HSC field name (used to pick FieldType + keywords) + value : the LLM-extracted value + metadata : dict from prompt_hsc_single_field; carries the + retrieved-chunks text and the LLM-stated confidence + contract_text : full cleaned contract text (for cross-checking) + +Output: + final score in [0.0, 1.0], rounded to 4 decimals. + +Method: + 1. Resolve FieldType -> weight preset. + 2. Compute each signal that has non-zero weight for that type. + Skip signals that return None (not applicable for this value). + 3. Renormalize weights over the signals that fired. + 4. Weighted average -> rule_score in [0, 1]. + 5. Blend lightly with the LLM-stated confidence as a smoothing prior. + (T3 will replace this prior with a calibrated LLM verifier.) +""" + +from __future__ import annotations + +import logging +import re +from typing import Optional + +from src.qc_qa.confidence import calibration as cal +from src.qc_qa.confidence import field_types as ft +from src.qc_qa.confidence import signals as sig + + +# How much weight to give the LLM-stated confidence in the final blend. +# Low on purpose: the rules dominate. T3 will replace this with a calibrated +# verifier. +_LLM_PRIOR_WEIGHT = 0.15 + +logger = logging.getLogger(__name__) + + +def _coerce_float(value: object) -> Optional[float]: + if value is None: + return None + try: + out = float(value) + except (TypeError, ValueError): + return None + if out != out: # NaN + return None + return max(0.0, min(1.0, out)) + + +_STOPWORDS = frozenset( + { + # Interrogatives / connectives we keep filtering out. + "what", + "where", + "when", + "which", + "whose", + "there", + "their", + "these", + "those", + "about", + "above", + "below", + "after", + "before", + "again", + "further", + "between", + "during", + "until", + "while", + "could", + "would", + "should", + "shall", + "might", + # Common short stopwords that slip past length >= 4 but carry no + # domain meaning, so they shouldn't contribute to na_check hits. + "this", + "that", + "with", + "from", + "into", + "have", + "your", + "more", + "only", + "such", + "than", + "they", + "them", + "also", + "each", + "very", + "some", + "most", + "many", + "much", + "into", + "been", + "were", + "will", + "must", + # Contract-boilerplate vocabulary. These appear in nearly every + # healthcare contract regardless of which field is being scored, + # so they're useless as na_check evidence and would otherwise + # flag every doc as suspicious. + "contract", + "contracts", + "contractual", + "agreement", + "agreements", + "document", + "documents", + "provider", + "providers", + "payer", + "payers", + "party", + "parties", + "service", + "services", + "section", + "sections", + "article", + "articles", + "paragraph", + "term", + "terms", + "covered", + # Prompt-instruction noise (these talk about the prompt, not the + # contract, so they're not useful evidence either way). + "value", + "field", + "given", + "named", + "answer", + "extract", + "extracted", + "return", + "follow", + "rules", + "guidelines", + "instructions", + "letters", + "numeric", + "alphanumeric", + "include", + "including", + "format", + "string", + "below", + } +) + +_KEYWORD_RE = re.compile(r"[A-Za-z][A-Za-z\-]{3,}") + + +def _field_keywords(field_name: str, field_prompt: str = "") -> list[str]: + """Keyword set for the na_check signal. + + Uses the field name as the keyword source. Field names like + CONTRACT_AMENDMENT_NUM are terse and domain-specific by design -- + split on underscores, drop boilerplate ("contract"), drop short + tokens ("num"), and what's left is the actual concept ("amendment"). + That's a far cleaner anti-hallucination signal than scanning a + verbose prompt that drags in generic English ("written", "letters", + "title", "numeric") that appears in every contract regardless. + + The prompt is a TRUE fallback: only consulted when the field name + yields nothing useful. For every HSC field today, the field name + produces at least one keyword, so the prompt never runs in practice. + Caps the result at 25 keywords. + """ + seen: dict[str, None] = {} + + def _add(token: str) -> bool: + """Add a normalized token; return True when the cap is reached.""" + tok = token.lower() + if len(tok) < 4 or tok in _STOPWORDS or tok in seen: + return False + seen[tok] = None + return len(seen) >= 25 + + # 1. Field name -- preferred source. + if field_name: + for tok in re.split(r"[_\W]+", field_name): + if _add(tok): + return list(seen.keys()) + + # 2. Prompt -- only as fallback when the field name had nothing + # useful to contribute (rare in practice). + if not seen and field_prompt: + for tok in _KEYWORD_RE.findall(field_prompt): + if _add(tok): + return list(seen.keys()) + + return list(seen.keys()) + + +def score_field( + field_name: str, + value: object, + metadata: dict | None, + contract_text: str = "", + field_prompt: str = "", +) -> float: + """Compute the rule-based confidence score for a single field. + + Always returns a finite float in [0.0, 1.0]. On any unexpected error, + falls back to the LLM-stated confidence (or 0.5) and logs a warning -- + a bad rule score must never crash the pipeline. + """ + metadata = metadata or {} + try: + field_type = ft.get_field_type(field_name) + weights = ft.get_weight_preset(field_type) + + retrieved_text = str(metadata.get("retrieved_chunks_text") or "") + keywords = _field_keywords(field_name, field_prompt) + + scores: dict[str, float] = {} + + if weights.get(ft.EXACT, 0) > 0: + v = sig.signal_exact(value, contract_text or retrieved_text) + if v is not None: + scores[ft.EXACT] = v + + if weights.get(ft.TOKEN_OVERLAP, 0) > 0: + v = sig.signal_token_overlap(value, contract_text or retrieved_text) + if v is not None: + scores[ft.TOKEN_OVERLAP] = v + + if weights.get(ft.FUZZY, 0) > 0: + # Fuzzy is O(n*m); keep it scoped to retrieved chunks when present. + haystack = retrieved_text or contract_text + v = sig.signal_fuzzy(value, haystack) + if v is not None: + scores[ft.FUZZY] = v + + if weights.get(ft.NUMBER_MATCH, 0) > 0: + v = sig.signal_number_match(value, contract_text or retrieved_text) + if v is not None: + scores[ft.NUMBER_MATCH] = v + + if weights.get(ft.REGEX, 0) > 0: + v = sig.signal_regex(value, field_type) + if v is not None: + scores[ft.REGEX] = v + + if weights.get(ft.GROUNDING, 0) > 0: + v = sig.signal_grounding(value, retrieved_text) + if v is not None: + scores[ft.GROUNDING] = v + + if weights.get(ft.NA_CHECK, 0) > 0: + v = sig.signal_na_check(value, contract_text, keywords) + if v is not None: + scores[ft.NA_CHECK] = v + + if not scores: + # No signal could fire -- fall back to the LLM's self-reported + # confidence (which we normalize) or a neutral 0.5. + return _coerce_float(metadata.get("confidence")) or 0.5 + + # Renormalize weights over the signals that actually fired. + total_weight = sum(weights[name] for name in scores) + if total_weight <= 0: + return _coerce_float(metadata.get("confidence")) or 0.5 + rule_score = sum(scores[n] * weights[n] for n in scores) / total_weight + + # Light blend with the LLM-stated confidence as a prior. The + # rule-based scorer dominates; the prior just smooths very-low + # rule scores when the LLM was very confident (and vice versa). + llm_conf = _coerce_float(metadata.get("confidence")) + if llm_conf is None: + final = rule_score + else: + final = (1 - _LLM_PRIOR_WEIGHT) * rule_score + _LLM_PRIOR_WEIGHT * llm_conf + + # Map raw -> empirically-calibrated confidence using the + # per-field-type curve. Identity by default; becomes meaningful + # once the curve is backfilled from QC data. + final = cal.calibrate(field_type, final) + + return round(max(0.0, min(1.0, final)), 4) + + except Exception as exc: # pragma: no cover - defensive + logger.warning( + "score_field failed for %s; falling back to LLM-stated conf: %s", + field_name, + exc, + ) + return _coerce_float((metadata or {}).get("confidence")) or 0.5 diff --git a/src/qc_qa/confidence/signals.py b/src/qc_qa/confidence/signals.py new file mode 100644 index 0000000..225735d --- /dev/null +++ b/src/qc_qa/confidence/signals.py @@ -0,0 +1,229 @@ +"""Rule-based confidence signals for one-to-one fields. + +Each signal returns a float in [0.0, 1.0] or None when not applicable. +None means "this signal could not produce evidence for this (value, context) +pair" -- the scorer skips it and renormalizes weights over the remaining +signals. + +Signals: + exact - normalized value is a substring of the context. + token_overlap - share of value tokens present in the context. + fuzzy - SequenceMatcher best-substring similarity. + number_match - all numbers in the value appear in the context. + regex - field-type-aware format check (date, money, TIN, etc.). + grounding - same family as exact/token but scoped to the *retrieved + chunks the LLM actually saw* (much stronger anti- + hallucination signal than scanning the whole contract). + na_check - special-case scoring when the value is N/A: rewards a + correctly-missing field, penalizes a suspicious miss. +""" + +from __future__ import annotations + +import re +from difflib import SequenceMatcher +from typing import Optional + +from src.qc_qa.confidence.field_types import FieldType + + +_NA_TOKENS = {"", "n/a", "na", "none", "null", "not found"} +_TOKEN_RE = re.compile(r"[A-Za-z0-9]+") +_NUMBER_RE = re.compile(r"\d+(?:[\.,]\d+)?") + +# Field-type-aware format regex. Matches the *value*, not the contract. +_FORMAT_PATTERNS = { + FieldType.DATE: re.compile( + r"\b\d{1,2}[/\-.]\d{1,2}[/\-.]\d{2,4}\b" + r"|\b\d{4}[/\-.]\d{1,2}[/\-.]\d{1,2}\b" + r"|\b(?:january|february|march|april|may|june|july|august|" + r"september|october|november|december)\b", + re.IGNORECASE, + ), + FieldType.MONEY: re.compile( + r"\$\s*[\d,]+(?:\.\d{1,2})?" + r"|\b\d{1,3}(?:,\d{3})+(?:\.\d+)?\b" + r"|\b\d+(?:\.\d+)?\s*(?:USD|dollars?)\b", + re.IGNORECASE, + ), + FieldType.TIN: re.compile(r"^\s*\d{2}-?\d{7}\s*$"), + FieldType.NPI: re.compile(r"^\s*\d{10}\s*$"), + FieldType.CODE: re.compile(r"^[A-Z0-9.\-]{1,12}$"), + FieldType.BOOLEAN: re.compile( + r"^\s*(?:y|n|yes|no|true|false|0|1)\s*$", re.IGNORECASE + ), +} + + +def _normalize(text: str) -> str: + """Lowercase + collapse whitespace. Used for substring/match comparisons.""" + return re.sub(r"\s+", " ", text.strip().lower()) + + +def is_na(value: object) -> bool: + """True when the extracted value should be treated as 'no answer'. + + Catches all the shapes a "missing" value can take coming out of the + pipeline: Python None, the standard N/A string tokens, AND pandas / + float NaN (which arrives as a float when the value column was empty + in the dataframe). Float NaN is the only float that is not equal to + itself, which is the cheapest portable check. + """ + if value is None: + return True + # Float NaN check before any str() coercion (str(float('nan')) == 'nan', + # which would otherwise slip past the token check below). + if isinstance(value, float) and value != value: + return True + if isinstance(value, (list, tuple, set)): + return all(is_na(item) for item in value) + return _normalize(str(value)) in _NA_TOKENS + + +def _stringify(value: object) -> str: + """Best-effort coercion of a value to a single string for matching.""" + if isinstance(value, (list, tuple, set)): + return " ".join(_stringify(v) for v in value) + return str(value) if value is not None else "" + + +# ----------------------------- signals ------------------------------------- + + +def signal_exact(value: object, context: str) -> Optional[float]: + """1.0 if the normalized value appears verbatim in the context, else 0.0.""" + if not context or is_na(value): + return None + needle = _normalize(_stringify(value)) + if not needle: + return None + return 1.0 if needle in _normalize(context) else 0.0 + + +def signal_token_overlap(value: object, context: str) -> Optional[float]: + """Share of value tokens (alphanumeric) present in the context.""" + if not context or is_na(value): + return None + value_tokens = [t.lower() for t in _TOKEN_RE.findall(_stringify(value))] + if not value_tokens: + return None + haystack_tokens = {t.lower() for t in _TOKEN_RE.findall(context)} + hits = sum(1 for t in value_tokens if t in haystack_tokens) + return hits / len(value_tokens) + + +def signal_fuzzy(value: object, context: str) -> Optional[float]: + """Best fuzzy substring similarity between the value and the context. + + Uses SequenceMatcher on the lowercased strings. Inexpensive on the + capped chunk text we feed it. For very long contracts we'd want a + sliding window, but for retrieved-chunks blobs the full match is fine. + """ + if not context or is_na(value): + return None + needle = _normalize(_stringify(value)) + if not needle: + return None + haystack = _normalize(context) + return SequenceMatcher(None, needle, haystack).ratio() + + +def signal_number_match(value: object, context: str) -> Optional[float]: + """Share of numbers in the value that also appear in the context. + + Useful for dates / money / TINs / NPIs where the "right answer" is a + set of digits. Returns None for values that don't contain any numbers + (the signal isn't applicable). + """ + if not context or is_na(value): + return None + value_numbers = _NUMBER_RE.findall(_stringify(value)) + if not value_numbers: + return None + haystack = context + hits = sum(1 for n in value_numbers if n in haystack) + return hits / len(value_numbers) + + +def signal_regex(value: object, field_type: FieldType) -> Optional[float]: + """Does the value match the format expected for its field type? + + Returns None for FREE_TEXT and NAME (no meaningful format check). + """ + if is_na(value): + return None + pattern = _FORMAT_PATTERNS.get(field_type) + if pattern is None: + return None + candidate = _stringify(value).strip() + return 1.0 if pattern.search(candidate) else 0.0 + + +def signal_grounding(value: object, retrieved_chunks_text: str) -> Optional[float]: + """How well does the value match the chunks the LLM actually saw? + + This is the strongest anti-hallucination signal: if the LLM produced + a value that isn't in its retrieved context, it likely made it up. + Combines exact substring (worth most) with fuzzy similarity (forgiving + on minor edits / capitalization / spacing). + """ + if not retrieved_chunks_text or is_na(value): + return None + needle = _normalize(_stringify(value)) + if not needle: + return None + haystack = _normalize(retrieved_chunks_text) + if needle in haystack: + return 1.0 + # Fall back to fuzzy: long haystack penalizes raw ratio, so use + # find_longest_match against a heuristic window. + fuzzy = SequenceMatcher(None, needle, haystack).ratio() + # Boost slightly when most value tokens are present even if not in order. + value_tokens = [t.lower() for t in _TOKEN_RE.findall(_stringify(value))] + if value_tokens: + haystack_tokens = {t.lower() for t in _TOKEN_RE.findall(haystack)} + token_hit_rate = sum(1 for t in value_tokens if t in haystack_tokens) / len( + value_tokens + ) + return max(fuzzy, 0.7 * token_hit_rate) + return fuzzy + + +def signal_na_check( + value: object, + contract_text: str, + field_keywords: Optional[list[str]] = None, +) -> Optional[float]: + """Score legitimate vs suspicious 'N/A' answers. + + Returns None for values that are NOT N/A (other signals cover those). + + For N/A values, counts total occurrences of the field's keywords in + the contract. A doc that genuinely contains the field's concept will + mention it many times; a doc that doesn't will have zero or near-zero + mentions. The score is continuous, not bucketed, so an amendment doc + that mentions "amendment" four times scores meaningfully lower (more + suspicious) than one that mentions it once in passing. + + Score ramp: + 0 hits -> 0.85 (the absence looks correct) + saturation -> 0.20 (the contract clearly discusses this; + the LLM's N/A is almost certainly wrong) + """ + if not is_na(value): + return None + if not contract_text or not field_keywords: + # Without keywords we can't tell suspicious from legitimate; stay + # neutral rather than guess. + return 0.6 + haystack = _normalize(contract_text) + # Count total occurrences (not just presence) so frequency carries + # signal. Saturation has a meaningful floor (~30 mentions) because + # contracts often contain a few mentions of any concept in boilerplate + # ("may be amended by written amendment", etc.) without actually being + # about it. Only docs that mention the concept tens of times are + # clearly "about" it; those are the ones whose N/A is suspicious. + total_hits = sum(haystack.count(kw.lower()) for kw in field_keywords if kw) + saturation = max(30, 5 * len(field_keywords)) + intensity = min(total_hits / saturation, 1.0) + return 0.85 - 0.65 * intensity diff --git a/src/qc_qa/confidence/summary.py b/src/qc_qa/confidence/summary.py new file mode 100644 index 0000000..72d1f7f --- /dev/null +++ b/src/qc_qa/confidence/summary.py @@ -0,0 +1,159 @@ +"""Batch-level confidence reporting. + +After all per-file 1:1 scoring is done, this module produces two CSVs: + + - CONFIDENCE-SUMMARY.csv : per-field distribution stats + columns: field, n, mean, p10, p50, p90, n_below_threshold, + threshold, pct_below_threshold + - CONFIDENCE-FLAGGED.csv : per-file rows where any field's confidence + landed below the threshold (lowest first) + columns: FILE_NAME, field, confidence, value, threshold + +Both are derived from the per-row _CONF columns the rule scorer +writes into the 1:1 results. + +Thresholds are read from config (default 0.6) and can be tuned per run +via the confidence_threshold CLI arg. +""" + +from __future__ import annotations + +import logging +from typing import Iterable, List, Tuple + +import numpy as np +import pandas as pd + +logger = logging.getLogger(__name__) + +CONF_SUFFIX = "_CONF" +DEFAULT_THRESHOLD = 0.6 + + +def _conf_columns(df: pd.DataFrame) -> List[str]: + return [c for c in df.columns if c.endswith(CONF_SUFFIX) and c != CONF_SUFFIX] + + +def _field_name_from_conf(col: str) -> str: + """Strip the _CONF suffix to get the underlying field name.""" + return col[: -len(CONF_SUFFIX)] + + +def _percentile(series: pd.Series, q: float) -> float: + """np.nanpercentile, but defensive against an all-NaN series.""" + arr = pd.to_numeric(series, errors="coerce").to_numpy() + arr = arr[~np.isnan(arr)] + if arr.size == 0: + return float("nan") + return float(np.percentile(arr, q)) + + +def compute_summary( + final_df: pd.DataFrame, threshold: float = DEFAULT_THRESHOLD +) -> pd.DataFrame: + """Build CONFIDENCE-SUMMARY.csv from the final results dataframe. + + Returns an empty DataFrame (correct columns) if there are no _CONF + columns to summarize — callers can skip writing rather than crashing. + """ + cols = [ + "field", + "n", + "mean", + "p10", + "p50", + "p90", + "n_below_threshold", + "threshold", + "pct_below_threshold", + ] + if final_df is None or final_df.empty: + return pd.DataFrame(columns=cols) + + conf_cols = _conf_columns(final_df) + if not conf_cols: + return pd.DataFrame(columns=cols) + + rows = [] + for col in conf_cols: + series = pd.to_numeric(final_df[col], errors="coerce").dropna() + n = int(series.shape[0]) + if n == 0: + rows.append( + { + "field": _field_name_from_conf(col), + "n": 0, + "mean": float("nan"), + "p10": float("nan"), + "p50": float("nan"), + "p90": float("nan"), + "n_below_threshold": 0, + "threshold": threshold, + "pct_below_threshold": float("nan"), + } + ) + continue + n_below = int((series < threshold).sum()) + rows.append( + { + "field": _field_name_from_conf(col), + "n": n, + "mean": round(float(series.mean()), 4), + "p10": round(_percentile(series, 10), 4), + "p50": round(_percentile(series, 50), 4), + "p90": round(_percentile(series, 90), 4), + "n_below_threshold": n_below, + "threshold": threshold, + "pct_below_threshold": round(100.0 * n_below / n, 2), + } + ) + + out = pd.DataFrame(rows, columns=cols) + out = out.sort_values(by="pct_below_threshold", ascending=False).reset_index( + drop=True + ) + return out + + +def compute_flagged( + final_df: pd.DataFrame, + threshold: float = DEFAULT_THRESHOLD, + file_name_column: str = "FILE_NAME", +) -> pd.DataFrame: + """Emit (FILE_NAME, field, confidence, value, threshold) rows for every + cell below the threshold. Reviewer hit list. + """ + cols = ["FILE_NAME", "field", "confidence", "value", "threshold"] + if final_df is None or final_df.empty: + return pd.DataFrame(columns=cols) + + conf_cols = _conf_columns(final_df) + if not conf_cols: + return pd.DataFrame(columns=cols) + + fn_col = file_name_column if file_name_column in final_df.columns else None + + flagged: List[dict] = [] + for col in conf_cols: + field = _field_name_from_conf(col) + series = pd.to_numeric(final_df[col], errors="coerce") + below_mask = series < threshold + if not below_mask.any(): + continue + below = final_df.loc[below_mask, [c for c in (fn_col, field) if c]] + for idx, row in below.iterrows(): + flagged.append( + { + "FILE_NAME": row[fn_col] if fn_col else "", + "field": field, + "confidence": round(float(series.loc[idx]), 4), + "value": "" if field not in below.columns else row.get(field, ""), + "threshold": threshold, + } + ) + + if not flagged: + return pd.DataFrame(columns=cols) + out = pd.DataFrame(flagged, columns=cols) + out = out.sort_values(by="confidence", ascending=True).reset_index(drop=True) + return out diff --git a/src/utils/io_utils.py b/src/utils/io_utils.py index 42f09b4..7093a8a 100644 --- a/src/utils/io_utils.py +++ b/src/utils/io_utils.py @@ -922,6 +922,28 @@ def write_local( quoting=1, ) return None + elif output_type == "confidence_summary": + output_dir = os.path.join( + config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp, "tracking" + ) + os.makedirs(output_dir, exist_ok=True) + df.to_csv( + os.path.join(output_dir, f"{config.BATCH_ID}-CONFIDENCE-SUMMARY.csv"), + index=False, + quoting=1, + ) + return None + elif output_type == "confidence_flagged": + output_dir = os.path.join( + config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp, "tracking" + ) + os.makedirs(output_dir, exist_ok=True) + df.to_csv( + os.path.join(output_dir, f"{config.BATCH_ID}-CONFIDENCE-FLAGGED.csv"), + index=False, + quoting=1, + ) + return None elif output_type == "dtc": output_dir = os.path.join(config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp) os.makedirs(output_dir, exist_ok=True) @@ -1135,6 +1157,18 @@ def write_s3( config.S3_CLIENT.put_object( Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer ) + elif output_type == "confidence_summary": + output_path = f"{config.BATCH_ID}/{run_timestamp}/tracking/{config.BATCH_ID}-CONFIDENCE-SUMMARY.csv" + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) + elif output_type == "confidence_flagged": + output_path = f"{config.BATCH_ID}/{run_timestamp}/tracking/{config.BATCH_ID}-CONFIDENCE-FLAGGED.csv" + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) elif output_type == "dtc": output_path = f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-DTC.csv" csv_buffer = df.to_csv(index=False).encode()