Updated src file structure
This commit is contained in:
Katon Minhas
committed by
Michael McGuinness
Michael McGuinness
parent
41bcd7f372
commit
b02d2cc448
@@ -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)
|
||||
|
||||
|
||||
|
||||
+4
-4
@@ -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
|
||||
|
||||
@@ -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 ##################
|
||||
|
||||
+2
-2
@@ -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__":
|
||||
|
||||
+9
-9
@@ -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}")
|
||||
|
||||
|
||||
+1
-4
@@ -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.
|
||||
"""
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user