From b02d2cc4484dd1839dd663af2ea49bd4c669e8f4 Mon Sep 17 00:00:00 2001 From: Katon Minhas Date: Wed, 26 Jun 2024 20:34:55 -0700 Subject: [PATCH] Updated src file structure --- src/{prompt_funcs.py => bottom_up_funcs.py} | 283 +++++------------- src/config.py | 8 +- src/file_processing.py | 9 +- src/main.py | 4 +- src/postprocess.py | 18 +- ...rocessingfuncs.py => postprocess_funcs.py} | 0 src/prompts.py | 5 +- src/top_down_funcs.py | 123 ++++++++ 8 files changed, 220 insertions(+), 230 deletions(-) rename src/{prompt_funcs.py => bottom_up_funcs.py} (51%) rename src/{postprocessingfuncs.py => postprocess_funcs.py} (100%) create mode 100644 src/top_down_funcs.py diff --git a/src/prompt_funcs.py b/src/bottom_up_funcs.py similarity index 51% rename from src/prompt_funcs.py rename to src/bottom_up_funcs.py index f06aeb5..f3b9957 100644 --- a/src/prompt_funcs.py +++ b/src/bottom_up_funcs.py @@ -1,91 +1,39 @@ -import json +import dict_operations +import postprocess_funcs +import prompts +import config +import claude_funcs + import difflib -import config -import prompts -import claude_funcs -import dict_operations -import utils -import postprocessingfuncs - -def get_least_similar(dict_list, original_dict, field): - # print(f"Running similarity for {field}") - - min_similarity = float('inf') - least_similar_dict = None - for dictionary in dict_list: - # Extract the methodology text - field_text = dictionary[field] - # print(f"Field text: {field_text}") - - # Compute similarity using difflib - similarity = difflib.SequenceMatcher(None, str(field_text), str(original_dict[field])).ratio() - # print(f"Simlarity: {similarity}") - - # If the similarity is less than the current minimum, update the minimum and the corresponding dictionary - if similarity < min_similarity: - min_similarity = similarity - least_similar_dict = dictionary - - return least_similar_dict - -def get_lesser_of_dict(dict_list, original_dict): - - unique_rates = list({d['REIMBURSEMENT_RATE'] for d in dict_list}) - unique_fees = list({d['REIMBURSEMENT_FLAT_FEE'] for d in dict_list}) - - # If the rates are different, return the lesser of dict with a different rate than original - if len(unique_rates) > 1: - least_similar_dict = get_least_similar(dict_list, original_dict, 'REIMBURSEMENT_RATE') - elif len(unique_fees) > 1: - least_similar_dict = get_least_similar(dict_list, original_dict, 'REIMBURSEMENT_FLAT_FEE') - else: - least_similar_dict = get_least_similar(dict_list, original_dict, 'FULL_METHODOLOGY') - - return least_similar_dict - - -def run_bottom_up_lesser(d, text_dict, tokens=4000): +def run_bottom_up(filename, text_dict): """ - Processes specific clauses or conditions within the document text, such as 'lesser of' or 'greater of', - using a language model to analyze and extract relevant data from the identified text segment. + Processes the text of a document using a two-tiered Bottom Up approach to extract key financial and operational information. - This function takes a dictionary representing initial results and a text dictionary, uses the page number from the initial - results to locate the relevant text, and generates a prompt for language model analysis. Based on the response, it - determines whether 'lesser of' or 'greater of' language is indicated, updates the results, and returns a new dictionary - for the updated data or retains the original if no such language is present. + This function first runs BOTTOM_UP_PRIMARY and BOTTOM_UP_SECONDARY prompts, then processes these initial results to further + refine and structure them into a dictionary form. Parameters: - d (dict): A dictionary containing initial processing results from previous analyses, including a 'page_num' key. + filename (str): The name of the file being processed, used to tag output data. text_dict (dict): A dictionary of text keyed by page numbers that contains the content to be analyzed. - tokens (int): Token limit for language model invocations, default set to 8000. Returns: - tuple: A tuple containing two dictionaries, the first for any identified 'lesser of' or 'greater of' data and the second - containing the original or updated data depending on the presence of such language. + list of dict: A list of dictionaries with each dictionary containing refined and structured information + from both primary and secondary Bottom Up analyses, all tagged with the filename. """ - prompt = prompts.BOTTOM_UP_LESSER(d, text_dict[d['page_num']]) - lesser_of_answer = claude_funcs.invoke_claude_3(prompt, max_tokens=tokens) - lesser_of_dict_list = dict_operations.primary_string_to_dict({d['page_num'] : lesser_of_answer}, d['Filename']) - # print(lesser_of_dict_list) - # If lesser of is Y - if len(lesser_of_dict_list) > 1: - lesser_of_dict = get_lesser_of_dict(lesser_of_dict_list, d) - else: - d['LESSER_OF_LANGUAGE_IND'] = 'N' - d['GREATER_OF_LANGUAGE_IND'] = 'N' - return ({}, d) + # Bottom Up Primary - SERVICE, REIMBURSEMENT_FLAT_FEE, REIMBURSEMENT_RATE, METHODOLOGY + answer_strings = run_bottom_up_primary(text_dict, 8000) + answer_dicts = dict_operations.primary_string_to_dict(answer_strings, filename) + answer_dicts_filtered = postprocess_funcs.filter_service_column(answer_dicts) - lesser_of_dict['page_num'] = d['page_num'] - if lesser_of_dict['LESSER_OF_LANGUAGE_IND'] == 'Y' or lesser_of_dict['GREATER_OF_LANGUAGE_IND'] == 'Y': - lesser_of_dict['SERVICE'] = d['SERVICE'] - return (lesser_of_dict, d) - else: - d['LESSER_OF_LANGUAGE_IND'] = 'N' - d['GREATER_OF_LANGUAGE_IND'] = 'N' - return ({}, d) + # Bottom Up Secondary + results_dicts = run_bottom_up_secondary(answer_dicts_filtered, text_dict, 8000) + + for d in results_dicts: + d['Filename'] = filename + return results_dicts # List of dictionaries def run_bottom_up_primary(text_dict, tokens): @@ -183,151 +131,72 @@ def run_bottom_up_secondary(answer_dicts, text_dict, tokens): return final_dicts -def run_bottom_up(filename, text_dict): - """ - Processes the text of a document using a two-tiered Bottom Up approach to extract key financial and operational information. - This function first runs BOTTOM_UP_PRIMARY and BOTTOM_UP_SECONDARY prompts, then processes these initial results to further - refine and structure them into a dictionary form. + +def run_bottom_up_lesser(d, text_dict, tokens=4000): + """ + Processes specific clauses or conditions within the document text, such as 'lesser of' or 'greater of', + using a language model to analyze and extract relevant data from the identified text segment. + + This function takes a dictionary representing initial results and a text dictionary, uses the page number from the initial + results to locate the relevant text, and generates a prompt for language model analysis. Based on the response, it + determines whether 'lesser of' or 'greater of' language is indicated, updates the results, and returns a new dictionary + for the updated data or retains the original if no such language is present. Parameters: - filename (str): The name of the file being processed, used to tag output data. + d (dict): A dictionary containing initial processing results from previous analyses, including a 'page_num' key. text_dict (dict): A dictionary of text keyed by page numbers that contains the content to be analyzed. + tokens (int): Token limit for language model invocations, default set to 8000. Returns: - list of dict: A list of dictionaries with each dictionary containing refined and structured information - from both primary and secondary Bottom Up analyses, all tagged with the filename. + tuple: A tuple containing two dictionaries, the first for any identified 'lesser of' or 'greater of' data and the second + containing the original or updated data depending on the presence of such language. """ - # Bottom Up Primary - SERVICE, REIMBURSEMENT_FLAT_FEE, REIMBURSEMENT_RATE, METHODOLOGY - answer_strings = run_bottom_up_primary(text_dict, 8000) - answer_dicts = dict_operations.primary_string_to_dict(answer_strings, filename) - answer_dicts_filtered = postprocessingfuncs.filter_service_column(answer_dicts) + def get_least_similar(dict_list, original_dict, field): + min_similarity = float('inf') + least_similar_dict = None + for dictionary in dict_list: + field_text = dictionary[field] + similarity = difflib.SequenceMatcher(None, str(field_text), str(original_dict[field])).ratio() + if similarity < min_similarity: + min_similarity = similarity + least_similar_dict = dictionary + return least_similar_dict - # Bottom Up Secondary - results_dicts = run_bottom_up_secondary(answer_dicts_filtered, text_dict, 8000) + def get_lesser_of_dict(dict_list, original_dict): + unique_rates = list({d['REIMBURSEMENT_RATE'] for d in dict_list}) + unique_fees = list({d['REIMBURSEMENT_FLAT_FEE'] for d in dict_list}) + if len(unique_rates) > 1: + least_similar_dict = get_least_similar(dict_list, original_dict, 'REIMBURSEMENT_RATE') + elif len(unique_fees) > 1: + least_similar_dict = get_least_similar(dict_list, original_dict, 'REIMBURSEMENT_FLAT_FEE') + else: + least_similar_dict = get_least_similar(dict_list, original_dict, 'FULL_METHODOLOGY') - for d in results_dicts: - d['Filename'] = filename - return results_dicts # List of dictionaries - + return least_similar_dict -def run_top_down_metal_level(d, page): - """ - Identifies and extracts the metal level of a contract within a specific line of business (LOB) from the provided page text. - This function determines whether the contract's LOB is associated with marketplace or commercial sectors by examining the - 'CONTRACT_LOB' key in the dictionary. If relevant, it constructs and sends a specific prompt to a language model to extract the - metal level (e.g., Bronze, Silver, Gold, Platinum). If the LOB is not relevant, it directly assigns 'N/A'. - - Parameters: - d (dict): A dictionary containing extracted information from primary Top Down analysis, specifically the 'CONTRACT_LOB' key. - page (str): The text content of the page that is being analyzed for metal level information. - - Returns: - str: The metal level of the contract as determined by the analysis, or 'N/A' if the contract LOB is not applicable. - """ - - if 'MARKETPLACE' in str(d['CONTRACT_LOB']).upper() or 'COMMERCIAL' in str(d['CONTRACT_LOB']).upper(): - prompt = prompts.TOP_DOWN_METAL_LEVEL(d['CONTRACT_LOB'], page) - answer = claude_funcs.invoke_claude_3(prompt, max_tokens=4000) + prompt = prompts.BOTTOM_UP_LESSER(d, text_dict[d['page_num']]) + lesser_of_answer = claude_funcs.invoke_claude_3(prompt, max_tokens=tokens) + lesser_of_dict_list = dict_operations.primary_string_to_dict({d['page_num'] : lesser_of_answer}, d['Filename']) + # print(lesser_of_dict_list) + # If lesser of is Y + if len(lesser_of_dict_list) > 1: + lesser_of_dict = get_lesser_of_dict(lesser_of_dict_list, d) else: - answer = 'N/A' - return answer - - -def run_top_down_date(type_, d, page): - """ - Extracts specific date-related information from a page of text using a Top Down processing approach. - - This function tailors the extraction to focus on either 'EFFECTIVE' or 'TERMINATION' dates by constructing - a prompt that directs a language model to search for and interpret date information relevant to the provided type. - It formats the input data to fit the processing needs, sends it along with the page text to the model, and captures - the model's response. - - Parameters: - type_ (str): Specifies the type of date to extract, 'EFFECTIVE' or 'TERMINATION'. - d (dict): A dictionary containing preliminary data extracted from the text, which may include metadata like filename or page number. - page (str): The text content of the page from which to extract the date. - - Returns: - str: The extracted date as a string, based on the model's interpretation of the input prompt and text context. - """ - formatted_d = utils.format_td_check([d], ['Filename', 'page_num']) - prompt = prompts.TOP_DOWN_DATE('EFFECTIVE', formatted_d, page) - answer = claude_funcs.invoke_claude_3(prompt, max_tokens=4000) - return answer - - -def top_down_secondary(td_results, text_dict): - """ - Conducts secondary processing on results obtained from the primary Top Down analysis of document text, enhancing detail and accuracy. - - This function iteratively enhances each dictionary result from the initial analysis by adding or refining information related to: - - Market metal levels by running a specific function to determine the contract marketplace metal level. - - Effective dates by analyzing the context around specific keywords in the text and extracting these dates. - - Termination dates, similarly extracted based on the presence and context of specific keywords. - - Parameters: - td_results (list of dict): Initial dictionaries from the primary Top Down processing containing basic extracted data. - text_dict (dict): Dictionary keyed by page numbers, providing the full text for corresponding analysis and extraction tasks. - - Returns: - list of dict: Updated list of dictionaries, each enhanced with additional detailed attributes like metal level and date specifics. - """ - updated_dicts = [] - for d in td_results: - # Run Metal Level - d['CONTRACT_MARKETPLACE_METAL_LEVEL'] = run_top_down_metal_level(d, text_dict[d['page_num']]) - - # Dates - d['LOB_PRICING_TERMS_EFFECTIVE_DATE'] = run_top_down_date('EFFECTIVE' , d, text_dict[d['page_num']]) - - # if auto_renewal == N: else: 'N/A' - d['LOB_PRICING_TERMS_TERMINATION_DATE'] = run_top_down_date('TERMINATION', d, text_dict[d['page_num']]) - - updated_dicts.append(d) - return updated_dicts - - - -def run_top_down(filename, text_dict): - """ - Executes the Top Down processing strategy on a provided dictionary of text pages, extracting structured data based on specified prompts. - - The function first filters and processes each page of text that is correctly numbered, using a primary analysis prompt. The results - from this analysis are then parsed into dictionaries, tagged with their respective page numbers and filename, and collected. - After the initial processing, a secondary Top Down analysis is conducted to refine and possibly expand the extracted data. - - Parameters: - filename (str): The filename associated with the text, used to tag the output data. - text_dict (dict): A dictionary where keys are page numbers and values are the text on those pages. - - Returns: - list of dict: A list of dictionaries where each dictionary contains structured results from both primary and secondary - Top Down analyses, associated with a specific page and the overall document. - """ - - all_results = [] - - # Primary - for page_num, page_text in text_dict.items(): - if not page_num.isdigit(): - continue - prompt = prompts.TOP_DOWN_PRIMARY(page_text) - answer = claude_funcs.invoke_claude_3(prompt, max_tokens=4000) - answer_dict = json.loads(answer) - answer_dict.update({'page_num' : page_num, 'Filename' : filename}) - #answer_dicts = dict_operations.primary_string_to_dict({page_num : answer}, filename) # Convert to dictionaries - all_results.append(answer_dict) - - # td_primary = [i for d in all_results for i in d] # Consolidate to one list - # print(td_primary) - - # Secondaries - td_final = top_down_secondary(all_results, text_dict) - - return td_final # List of list of dictionaries + d['LESSER_OF_LANGUAGE_IND'] = 'N' + d['GREATER_OF_LANGUAGE_IND'] = 'N' + return ({}, d) + + lesser_of_dict['page_num'] = d['page_num'] + if lesser_of_dict['LESSER_OF_LANGUAGE_IND'] == 'Y' or lesser_of_dict['GREATER_OF_LANGUAGE_IND'] == 'Y': + lesser_of_dict['SERVICE'] = d['SERVICE'] + return (lesser_of_dict, d) + else: + d['LESSER_OF_LANGUAGE_IND'] = 'N' + d['GREATER_OF_LANGUAGE_IND'] = 'N' + return ({}, d) diff --git a/src/config.py b/src/config.py index fa341aa..c0b8b3b 100644 --- a/src/config.py +++ b/src/config.py @@ -23,7 +23,7 @@ READ_MODE = '_LOCAL_' # OR '_S3_' WRITE_OUTPUT = True # True writes csvs, False prints result in console but no output written OUTPUT_DIRECTORY = 'output' CONSOLIDATED_OUTPUT_DIRECTORY = 'output_consolidated' -OUTPUT_CSV_PATH = f'consolidated_output_{TODAY}.xlsx' +OUTPUT_CSV_PATH = f'consolidated_output_{TODAY}.csv' TD_RESULTS_NAME = 'td_results.csv' BU_RESULTS_NAME = 'bu_results.csv' UNPROCESSED_RESULTS_NAME = 'combined_results_unprocessed.csv' @@ -54,9 +54,9 @@ VALID_COLUMNS = ['Filename', 'page_num', 'SERVICE', 'REIMBURSEMENT_FLAT_FEE', 'R 'Corrected_LOB', 'Corrected_PROGRAM', 'Corrected_NETWORK'] # AWS Keys -AWS_ACCESS_KEY_ID="ASIAZTMXAXNXDIZUYWWF" -AWS_SECRET_ACCESS_KEY="OFmTz8zdH+d63adfyKUXqefv9hH31K8YryJMWv8v" -AWS_SESSION_TOKEN="IQoJb3JpZ2luX2VjEP///////////wEaCXVzLWVhc3QtMiJIMEYCIQDfd7FVLKfCelPBzBlulypHwvAT+Qe3/Z9PFy/5U+rdDgIhAJ9kXpN4c16rmv26mnKlSQ+mMNqKzz0g8gk+zvHUbsxKKowDCKj//////////wEQABoMNjYwMTMxMDY4NzgyIgyOxND9AUVES8laaowq4ALruIJOnTW8Wb1DODUG/ftp8Zr/J9tnqZX8gmTeFK6v1/h++iWTwvD2gdX/MIKDNBL7acFV7DqeffShDH8rdSR2KsGplsZUMkeu/nzt9TSvE69OvyTwIb/Jw45q3PCnnClytwZIuW2kcxawsDHUO/nEVr/e1Y6PyvHs4fhaC5+6BlIjVqpCK3bM4Pk3K+EgBbZkRXSqm3zEAwFYBbRnXONBHWSrpADGxhwJkqo8yUiyBoDysl8rDQTeATrwDawU9vmEmbRJH4XUHxdtOWixUhbzwyTnB1QW9lMdi6BL+GzSGj4XLZWzJj4EGW32g70yamnb6JyV9Ju7Uzsbrusz1Qxg9/apmGmP4fAigzV3r1chv/Vb8Lukm+ljB82+ZqSyiy5bukdAfDoObjdxe/BGLPGkLVN9bn3j29lJs8+tNjfmlTr4boXEj3s72KRwSyEpIw+Tr/MY9yP/mkf+vzti2l6rMKuO5rMGOqUBvP5K7JUpoUMIx2zhz+Legs1pDzcnhQ3Hc0wcd2QqS3crnENTGIFKNEJ/XIbDwTwcnIiiJYsPxKfedv8H1ZBFFNA7sNWggPtlPolKLyxZHfs3RZMsp+MOvR55fgyvC14REfALGRcgqNHsoYI61jrW715rSMI8di7Qv8MvSXICVNUI6+hGAPqs9tLeJBHnaEcclOqR5/YQ5rNq5gnQvD7117MSW5Sk" +AWS_ACCESS_KEY_ID="ASIAZTMXAXNXBHQKW32L" +AWS_SECRET_ACCESS_KEY="IozsUqPsyXu2klzNMU42CDZ1ZtGz8Yo29mhuJRaQ" +AWS_SESSION_TOKEN="IQoJb3JpZ2luX2VjEDsaCXVzLWVhc3QtMiJHMEUCIFkt/LwV1X3UmAyNi25xrDIL+mdaV7eV3U03VUQPc0o/AiEA4R54Gyd4d/55N5P0a84zzHiPUuejjyEMF6ARAk0i2fgqjAMI5P//////////ARAAGgw2NjAxMzEwNjg3ODIiDC+Kl0HOGnW+Vct/FyrgAg+t9ssFLXSj0VFXfVud6jLJnIAChoXSikj59wmU1hX3vcsN699t4UT5vfAjIwZ3emEPrfaKnuvcdC4ev2V4fT3IxRS8FLu6eUKqImF8ZkATIxg0nYpPqx+92CBvcT1gjLv4Y/1oorb0GxLcTiZLBY6IrCbrr3bDyexU+fgqoGlVFhaqfVPotvXy5+s0RbFWF7wQ0IanZ4eX5ZTFUXZMJh0LBYSTIuas6oDF2z/k/De48NKOWEZCmVxIG/ykyIbBujSICetKKX1qGtB74NrxSfhERGfof4DjZl5q4SUEouTRVb5aGk779lqdKjTXjmGyWyryzUu9jJfqyZiC+ymcIliK4aTn4W88/prqx3p1mq+/T1e5uMhtXi/9JURT3P/avOVz34j5ohggUxqvl2TBASKaUGLrOIBoJzoMevgodoWy1TleJe7xJWBOS5nq5knxMOreHjuImgNofmiRuSNbayowpqTzswY6pgFwPrVOCxqQyxXwNsCEB27me72Ez3splxi+JYiYyaxkANoftb6d1/JyDBprnsact7hvD3mPVUhjdvsh92o5+F1lR0usiyCqrvUrXgpYKdBh6jIJh2kcNDSYwoGMHI2kVb2G53GWe1Vb06cS79Ei3QVurxVe7SVYVqd+bpicuYDdcrvdzCeF2E+9P9qEQa1/4B6ei+0O+lao4WBjvy7ZyuAlwaUUgsw0" # File Paths LOCAL_PATH = 'data/test/' # Replace with local diff --git a/src/file_processing.py b/src/file_processing.py index 4880f45..de67684 100644 --- a/src/file_processing.py +++ b/src/file_processing.py @@ -5,8 +5,9 @@ import pandas as pd import config import preprocess import table_funcs -import prompt_funcs -import postprocessingfuncs +import top_down_funcs +import bottom_up_funcs +import postprocess_funcs import postprocess import merge_funcs @@ -48,11 +49,11 @@ def process_file(file_object): text_dict = preprocess.highlight_rates(text_dict) ################## RUN TOP DOWN ################## - td_results = prompt_funcs.run_top_down(filename, text_dict) # Returns list of dictionaries for each page + td_results = top_down_funcs.run_top_down(filename, text_dict) # Returns list of dictionaries for each page print(f"Top Down Complete - {filename}") ################## RUN BOTTOM UP ################## - bu_results = prompt_funcs.run_bottom_up(filename, text_dict) # Returns list of dictionaries + bu_results = bottom_up_funcs.run_bottom_up(filename, text_dict) # Returns list of dictionaries print(f"Bottom Up Complete - {filename}") ################## COMBINE TOP DOWN AND BOTTOM UP ################## diff --git a/src/main.py b/src/main.py index 43ec3f8..b61c130 100644 --- a/src/main.py +++ b/src/main.py @@ -12,9 +12,8 @@ Functional Overview: """ # Imports -import concurrent.futures -import traceback import pandas as pd +import os import utils import config @@ -38,6 +37,7 @@ def main(): # Write Consolidated Output if config.WRITE_OUTPUT: all_results_df = pd.concat(all_results).reset_index(drop=True) + all_results_df.to_csv(os.path.join(config.CONSOLIDATED_OUTPUT_DIRECTORY, config.OUTPUT_CSV_PATH)) if __name__ == "__main__": diff --git a/src/postprocess.py b/src/postprocess.py index e28b150..efa3039 100644 --- a/src/postprocess.py +++ b/src/postprocess.py @@ -1,7 +1,7 @@ import pandas as pd import re -import postprocessingfuncs +import postprocess_funcs import config @@ -16,49 +16,49 @@ def postprocess_results(combined_df): # Sanitize the combined data try: - df = postprocessingfuncs.sanitize_combined(combined_df) + df = postprocess_funcs.sanitize_combined(combined_df) except Exception as e: print(f"Postprocessing Error - santize_combined : {e}") # Clean specific columns for exact matches for field_list in [['CONTRACT_LOB', 'Corrected_LOB', config.VALID_LOBS], ['CONTRACT_PROGRAM', 'Corrected_PROGRAM', config.VALID_PROGRAMS], ['CONTRACT_NETWORK', 'Corrected_NETWORK', config.VALID_NETWORKS]]: try: - postprocessingfuncs.clean_columns_combined(df, field_list[0], field_list[2], field_list[1]) + postprocess_funcs.clean_columns_combined(df, field_list[0], field_list[2], field_list[1]) except Exception as e: print(f"Postprocessing Error - clean_columns_combined : {e}") try: - postprocessingfuncs.clean_columns_combined_fuzzy(df, field_list[0], field_list[2], config.FUZZY_MATCH_THRESHOLD) + postprocess_funcs.clean_columns_combined_fuzzy(df, field_list[0], field_list[2], config.FUZZY_MATCH_THRESHOLD) except Exception as e: print(f"Postprocessing Error - clean_columns_combined_fuzzy : {e}") # Correct misplaced values across columns try: - df = postprocessingfuncs.correct_misplaced_values(df, ['CONTRACT_LOB', 'CONTRACT_PROGRAM', 'CONTRACT_NETWORK'], valid_values_dict) + df = postprocess_funcs.correct_misplaced_values(df, ['CONTRACT_LOB', 'CONTRACT_PROGRAM', 'CONTRACT_NETWORK'], valid_values_dict) except Exception as e: print(f"Postprocessing Error - correct_misplaced_values : {e}") # Move percentages and large numbers to correct columns try: - df = postprocessingfuncs.move_percentage_to_rate(df) + df = postprocess_funcs.move_percentage_to_rate(df) except Exception as e: print(f"Postprocessing Error - move_percentage_to_rate : {e}") # Replace N/As in "Not Covered" with 0% try: - df = postprocessingfuncs.set_rate_to_zero_if_not_covered(df) + df = postprocess_funcs.set_rate_to_zero_if_not_covered(df) except Exception as e: print(f"Postprocessing Error - set_rate_to_zero_if_not_covered : {e}") # Adjust for adding or subtracting try: - df = postprocessingfuncs.adjust_reimbursement_rate(df) + df = postprocess_funcs.adjust_reimbursement_rate(df) except Exception as e: print(f"Postprocessing Error - adjust_reimbursement_rate : {e}") # Clean dates so that Termination and Effective aren't the same try: - df = postprocessingfuncs.clean_dates(df) + df = postprocess_funcs.clean_dates(df) except Exception as e: print(f"Postprocessing Error - clean_dates : {e}") diff --git a/src/postprocessingfuncs.py b/src/postprocess_funcs.py similarity index 100% rename from src/postprocessingfuncs.py rename to src/postprocess_funcs.py diff --git a/src/prompts.py b/src/prompts.py index 33fabf9..f174602 100644 --- a/src/prompts.py +++ b/src/prompts.py @@ -110,14 +110,11 @@ The above text contains a number of reimbursement terms. For the rest of this re Service: {d['SERVICE']} Methodology: {d['FULL_METHODOLOGY']} -Your job is to identify if the listed reimbursement contains some sort of exception (like for certain codes, hospitals, or other situations) or an escalation +Your job is to identify if the listed reimbursement contains some sort of exception (like for certain codes, hospitals, or other situations). Read and analyze the page, then populate a JSON dictionary with the following fields (descriptions provided): REIMBURSEMENT_EXCEPTION_IND: If the listed reimbursement contains some sort of exception (like for certain codes, hospitals, or other situations), return 'Y'. Otherwise, 'N' REIMBURSEMENT_DESCRIBE_EXCEPTION: If the listed reimbursement contains some sort of exception, describe it here. Try to use exact words from the text only. If there is none, just write 'N/A' -RATE_ESCALATOR_IND: Does the listed reimbursement include annual rate escalators for year-over-year rate increases? If yes, return 'Y'. Otherwise, 'N'. -RATE_ESCALATOR_BASIS: If contract does include annual rate escalators, on what are they based? Do not be overly verbose in your answer. -RATE_ESCALATOR_YEARLY_PERCENT_INCREASE: What is the percent value of increase or decrease year-over-year? If a decrease, return a negative percentage. Only return the dictionary, with no other commentary or explanation. Ensure you abide by proper JSON formatting. """ diff --git a/src/top_down_funcs.py b/src/top_down_funcs.py new file mode 100644 index 0000000..cbfd6a6 --- /dev/null +++ b/src/top_down_funcs.py @@ -0,0 +1,123 @@ + +import prompts +import claude_funcs +import utils + +import json + +def run_top_down_metal_level(d, page): + """ + Identifies and extracts the metal level of a contract within a specific line of business (LOB) from the provided page text. + + This function determines whether the contract's LOB is associated with marketplace or commercial sectors by examining the + 'CONTRACT_LOB' key in the dictionary. If relevant, it constructs and sends a specific prompt to a language model to extract the + metal level (e.g., Bronze, Silver, Gold, Platinum). If the LOB is not relevant, it directly assigns 'N/A'. + + Parameters: + d (dict): A dictionary containing extracted information from primary Top Down analysis, specifically the 'CONTRACT_LOB' key. + page (str): The text content of the page that is being analyzed for metal level information. + + Returns: + str: The metal level of the contract as determined by the analysis, or 'N/A' if the contract LOB is not applicable. + """ + + if 'MARKETPLACE' in str(d['CONTRACT_LOB']).upper() or 'COMMERCIAL' in str(d['CONTRACT_LOB']).upper(): + prompt = prompts.TOP_DOWN_METAL_LEVEL(d['CONTRACT_LOB'], page) + answer = claude_funcs.invoke_claude_3(prompt, max_tokens=4000) + else: + answer = 'N/A' + return answer + + +def run_top_down_date(type_, d, page): + """ + Extracts specific date-related information from a page of text using a Top Down processing approach. + + This function tailors the extraction to focus on either 'EFFECTIVE' or 'TERMINATION' dates by constructing + a prompt that directs a language model to search for and interpret date information relevant to the provided type. + It formats the input data to fit the processing needs, sends it along with the page text to the model, and captures + the model's response. + + Parameters: + type_ (str): Specifies the type of date to extract, 'EFFECTIVE' or 'TERMINATION'. + d (dict): A dictionary containing preliminary data extracted from the text, which may include metadata like filename or page number. + page (str): The text content of the page from which to extract the date. + + Returns: + str: The extracted date as a string, based on the model's interpretation of the input prompt and text context. + """ + formatted_d = utils.format_td_check([d], ['Filename', 'page_num']) + prompt = prompts.TOP_DOWN_DATE('EFFECTIVE', formatted_d, page) + answer = claude_funcs.invoke_claude_3(prompt, max_tokens=4000) + return answer + + +def top_down_secondary(td_results, text_dict): + """ + Conducts secondary processing on results obtained from the primary Top Down analysis of document text, enhancing detail and accuracy. + + This function iteratively enhances each dictionary result from the initial analysis by adding or refining information related to: + - Market metal levels by running a specific function to determine the contract marketplace metal level. + - Effective dates by analyzing the context around specific keywords in the text and extracting these dates. + - Termination dates, similarly extracted based on the presence and context of specific keywords. + + Parameters: + td_results (list of dict): Initial dictionaries from the primary Top Down processing containing basic extracted data. + text_dict (dict): Dictionary keyed by page numbers, providing the full text for corresponding analysis and extraction tasks. + + Returns: + list of dict: Updated list of dictionaries, each enhanced with additional detailed attributes like metal level and date specifics. + """ + updated_dicts = [] + for d in td_results: + # Run Metal Level + d['CONTRACT_MARKETPLACE_METAL_LEVEL'] = run_top_down_metal_level(d, text_dict[d['page_num']]) + + # Dates + d['LOB_PRICING_TERMS_EFFECTIVE_DATE'] = run_top_down_date('EFFECTIVE' , d, text_dict[d['page_num']]) + + # if auto_renewal == N: else: 'N/A' + d['LOB_PRICING_TERMS_TERMINATION_DATE'] = run_top_down_date('TERMINATION', d, text_dict[d['page_num']]) + + updated_dicts.append(d) + return updated_dicts + + + +def run_top_down(filename, text_dict): + """ + Executes the Top Down processing strategy on a provided dictionary of text pages, extracting structured data based on specified prompts. + + The function first filters and processes each page of text that is correctly numbered, using a primary analysis prompt. The results + from this analysis are then parsed into dictionaries, tagged with their respective page numbers and filename, and collected. + After the initial processing, a secondary Top Down analysis is conducted to refine and possibly expand the extracted data. + + Parameters: + filename (str): The filename associated with the text, used to tag the output data. + text_dict (dict): A dictionary where keys are page numbers and values are the text on those pages. + + Returns: + list of dict: A list of dictionaries where each dictionary contains structured results from both primary and secondary + Top Down analyses, associated with a specific page and the overall document. + """ + + all_results = [] + + # Primary + for page_num, page_text in text_dict.items(): + if not page_num.isdigit(): + continue + prompt = prompts.TOP_DOWN_PRIMARY(page_text) + answer = claude_funcs.invoke_claude_3(prompt, max_tokens=4000) + answer_dict = json.loads(answer) + answer_dict.update({'page_num' : page_num, 'Filename' : filename}) + #answer_dicts = dict_operations.primary_string_to_dict({page_num : answer}, filename) # Convert to dictionaries + all_results.append(answer_dict) + + # td_primary = [i for d in all_results for i in d] # Consolidate to one list + # print(td_primary) + + # Secondaries + td_final = top_down_secondary(all_results, text_dict) + + return td_final # List of list of dictionaries \ No newline at end of file