5/2 test update

src/bottom_up_funcs.py

@@ -24,17 +24,15 @@ def run_bottom_up_lesser(d, text_dict, tokens=8000):
         return ({}, d)
 
 
-
 def run_bottom_up_primary(text_dict, tokens):
     #chunk_dict = preprocess.chunk_text(text_dict)
-    if config.VERBOSE: print("Running Bottom Up Primary")
-
     answer_dict = {}
     #for page_number in chunk_dict.keys():
         #if '%' in chunk_dict[page_number] or '$' in chunk_dict[page_number]:
             #prompt = prompts.BOTTOM_UP_PRIMARY(chunk_dict[page_number], config.CLIENT_NAME)
     for page_number in text_dict.keys():
         if  page_number.isdigit() and ('%' in text_dict[page_number] or '$' in text_dict[page_number]):
+            # Run Primary
             prompt = prompts.BOTTOM_UP_PRIMARY(text_dict[page_number], config.CLIENT_NAME)
             answer = claude_funcs.invoke_claude_3(prompt, max_tokens=tokens)
             answer_dict[page_number] = answer
@@ -42,17 +40,9 @@ def run_bottom_up_primary(text_dict, tokens):
 
 
 def run_bottom_up_secondary(answer_dicts, text_dict, tokens):
-    if config.VERBOSE: print("Running Bottom Up Secondary")
-
     # New rows for lesser of
     temp_dicts = []
     for d in answer_dicts:     
-        # Bottom Up LOB
-        # prompt = prompts.BOTTOM_UP_LOB(dict, text_dict[d['page_num']])
-        # lob_answer = claude_funcs.invoke_claude_3(prompt, max_tokens=tokens)
-        # lob_dict = dict_operations.dict_string_to_dict(lob_answer)
-        # d = d.update(lob_dict)
-
         # Bottom Up Lesser
         lesser_object = run_bottom_up_lesser(d.copy(), text_dict.copy(), tokens)
         temp_dicts.append(lesser_object[1]) # Add original object from d
@@ -63,10 +53,10 @@ def run_bottom_up_secondary(answer_dicts, text_dict, tokens):
     for d in temp_dicts:
         if d is not None:
             page_num = d['page_num']
-            # # Bottom Up Methodology
-            # prompt = prompts.BOTTOM_UP_METHODOLOGY(dict)
-            # methodology_answer = claude_funcs.invoke_claude_3(prompt, max_tokens=tokens)
-            # dict['REIMBURSEMENT_METHODOLOGY'] = methodology_answer
+            # Bottom Up Methodology
+            prompt = prompts.BOTTOM_UP_METHODOLOGY(d)
+            methodology_answer = claude_funcs.invoke_claude_3(prompt, max_tokens=100)
+            d['REIMBURSEMENT_METHODOLOGY'] = methodology_answer
 
             # # Bottom Up FS
             prompt = prompts.BOTTOM_UP_FS(d)
@@ -90,6 +80,28 @@ def run_bottom_up_secondary(answer_dicts, text_dict, tokens):
         
     return final_dicts
 
+def run_bottom_up_lob(answer_dicts, text_dict, tokens):
+    lob_dict = {}
+    for page_num in set(d['page_num'] for d in answer_dicts):
+        try:
+            text_section = text_dict[str(int(page_num)-1)] + text_dict[page_num]
+        except:
+            text_section = text_dict[page_num]
+        prompt = prompts.BOTTOM_UP_LOB(text_section)
+        answer = claude_funcs.invoke_claude_3(prompt, max_tokens=tokens)
+        answer_dict = dict_operations.secondary_string_to_dict(answer)
+        lob_dict[page_num] = answer_dict
+    return lob_dict
+
+def consolidate_lob(lob_dicts, results_dicts):
+    final_dicts = []
+    for d in results_dicts:
+        try:
+            d.update(lob_dicts[d['page_num']])
+            final_dicts.append(d)
+        except:
+            final_dicts.append(d)
+    return final_dicts
 
 def bottom_up(file_object):
     #### PREPROCESS ####
@@ -105,16 +117,22 @@ def bottom_up(file_object):
     answer_dicts = dict_operations.primary_string_to_dict(answer_strings)
     answer_dicts = postprocess.primary_filter(answer_dicts)
 
+    # Bottom Up LOB - CONTRACT_LOB, CONTRACT_PROGRAM, CONTRACT_PRODUCT, PROV_TYPE
+    lob_dicts = run_bottom_up_lob(answer_dicts, text_dict, 4000)
+
     # Bottom Up Secondary
     results_dicts = run_bottom_up_secondary(answer_dicts, text_dict, 8000)
 
+    # Append LOB
+    final_dicts = consolidate_lob(lob_dicts, results_dicts)
+
     #### POSTPROCESS ####
     #answer_dicts = append_secondary(answer_dicts)
 
-    for dict in results_dicts:
-        dict['Filename'] = filename
+    for d in final_dicts:
+        d['Filename'] = filename
 
-    answer_df = pd.DataFrame(results_dicts)
+    answer_df = pd.DataFrame(final_dicts)
 
     # Write output
     if config.WRITE_OUTPUT:

src/bottom_up_main.py

@@ -1,6 +1,7 @@
 # Imports
 import os
 import concurrent.futures
+import traceback
 
 import utils
 import config
@@ -13,6 +14,8 @@ def main():
     else:
         # Read contract txt
         input_dict = utils.read_input()
+        already_processed = [s.split('.txt_results')[0]+'.txt' for s in os.listdir('results/')]
+        input_dict = {key : input_dict[key] for key in input_dict.keys() if key not in already_processed}
 
         # Set up results folder
         if not os.path.exists('results'):
@@ -22,10 +25,31 @@ def main():
         #tables_dict = table_funcs.extract_all_tables(input_dict)
         #print(tables_dict)
 
-        # Run bottom up primary with multithreading
+        # # Run bottom up primary with multithreading
         # with concurrent.futures.ThreadPoolExecutor(max_workers=config.MAX_WORKERS) as executor:
         #     executor.map(bottom_up_funcs.bottom_up, input_dict.items())
-        bottom_up_funcs.bottom_up(list(input_dict.items())[0])
+        
+        def process_item(item):
+            key, value = item
+            try:
+                bottom_up_funcs.bottom_up(item)
+            except Exception as e:
+                # Print the error message and traceback
+                print(f"Error processing item {key}: {e}")
+                traceback.print_exc()
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=config.MAX_WORKERS) as executor:
+            # Process each item individually
+            futures = [executor.submit(process_item, item) for item in input_dict.items()]
+
+            # Wait for all futures to complete
+            for future in concurrent.futures.as_completed(futures):
+                # Retrieve any exceptions raised by the thread
+                try:
+                    future.result()
+                except Exception as e:
+                    print(f"Error: {e}")
+
 
         # Write Output
         if config.WRITE_OUTPUT and config.OUTPUT_MODE == '_CONSOLIDATED_':

src/config.py

@@ -17,9 +17,12 @@ OUTPUT_MODE = '_INDIVIDUAL_'  # or'_CONSOLIDATED_'
 MAX_WORKERS = 20
 
 # AWS Keys
+AWS_ACCESS_KEY_ID="update here"
+AWS_SECRET_ACCESS_KEY="update here"
+AWS_SESSION_TOKEN="update here"
 
 # File Paths
-LOCAL_PATH = 'docs/test/' # Replace with local
+LOCAL_PATH = 'docs/all_txt_updated/' # Replace with local
 
 # S3 Settings
 S3_CLIENT = boto3.client('s3',
@@ -41,4 +44,3 @@ BEDROCK_RUNTIME = boto3.client(service_name="bedrock-runtime",
 MODEL_ID_CLAUDE3_HAIKU = 'anthropic.claude-3-haiku-20240307-v1:0'
 MODEL_ID_CLAUDE3_SONNET = 'anthropic.claude-3-sonnet-20240229-v1:0'
 MODEL_ID_CLAUDE2 = 'anthropic.claude-instant-v1'
-

src/dict_operations.py

@@ -2,6 +2,21 @@ import re
 import json
 from collections import defaultdict
 
+def secondary_string_to_dict(dict_string):
+    dict_string = dict_string.replace('<<<', '')
+    dict_string = dict_string.replace('>>>', '')
+    dict_string = dict_string.replace('\n', '') # Remove new line
+
+    start_index = dict_string.find('{')
+    end_index = dict_string.rfind('}') + 1
+    dict_substring = dict_string[start_index:end_index]
+    try:
+        result_dict = json.loads(dict_substring)
+    except:
+        dict_substring = dict_substring.replace("'", '"')
+        result_dict = json.loads(dict_substring)   
+    return result_dict
+
 def primary_string_to_dict(string_dict):
     data = []
     pattern = r'\{.*?\}'
@@ -10,22 +25,16 @@ def primary_string_to_dict(string_dict):
         dicts = primary_list.split('[')[1] # Strip front
         dicts = dicts.split(']')[0] # Strip back
         dicts = dicts.replace('\n', '') # Remove new lines
+        dicts = dicts.replace('<<<', '')
+        dicts = dicts.replace('>>>', '')
         dicts = re.sub(r"(?<!\\)'([^']*?)'(?<!\\):", r'"\1":', dicts)
         dict_list = re.findall(pattern, dicts)
-        dict_list = [json.loads(d) for d in dict_list]
+        dict_list = [secondary_string_to_dict(d) for d in dict_list]
         for dict_ in dict_list:
             dict_['page_num'] = page_num
             data.append(dict_)
     return data
 
-def secondary_string_to_dict(dict_string):
-    start_index = dict_string.find('{')
-    end_index = dict_string.rfind('}') + 1
-    dict_substring = dict_string[start_index:end_index]
-    dict_substring = dict_substring.replace("'", '"')
-    result_dict = json.loads(dict_substring)
-    return result_dict
-
 def append_secondary(result_dicts): 
     # Helper function to create a key from dict excluding certain keys
     def create_key(d):

src/postprocess.py

@@ -2,7 +2,6 @@ import config
 
 # Filter applied between primary and secondary
 def primary_filter(answer_dict):
-    if config.VERBOSE: print("Applying filter")
     filtered_dicts = []
     for d in answer_dict:
         if 'SERVICE' in d.keys():

src/prompts.py

@@ -62,21 +62,24 @@ Here are the attributes to be included in each dictionary, and instructions on h
 Only return the list, with no other commentary or explanation. Ensure you abide by proper JSON formatting.
 """
 
-def BOTTOM_UP_LOB(d, page):
+def BOTTOM_UP_LOB(page):
     return f"""### PAGE START ### {page} ### PAGE END
 
-The above text is one page of a contract. The page states that {d['SERVICE']} is reimbursed with the following methodology: {d['FULL_METHODOLOGY']}.
+The above text represents two pages from a contract between a healthcare Payer and Provider.
 
-Your job is to identify attributes related to this reimbursement term and return them in a JSON-formatted dictionary.
+Your job is to identify what LOBs, Products, Plans, and Provider Types the reimbursement terms on this page apply to.
 
-If any of the attributes are not found, return N/A. For all attributes, only write what is written on the page. Do not make up new phrases or words.
-
- Return a dictionary object with the following attributes:
+Return a dictionary object with the following attributes:
 'CONTRACT_LOB' : What Line of Business is the reimbursement schedule for? It will be Medicaid, Medicare, Marketplace, or similar. It will likely only be listed once per page, if at all.
 'CONTRACT_PROGRAM' : What Program is the reimbursement schedule for? This may be something like CHIP, CHIP PERINATE, STAR, STAR PLUS, or similar. This will likely only be listed once per page, if at all.
-'PRODUCT' : What Product is the reimbursement schedule for? This the brand name of the health plan. It is NOT the name of a hospital. This will likely only be listed once per page, if at all.
+'PRODUCT' : What Product is the reimbursement schedule for? This the brand name of the health plan. It is NOT the same as LOB. It is NOT the name of a hospital. This will likely only be listed once per page, if at all.
 'PROV_TYPE' : For what provider type or place of service is this fee schedule for? It could be Hospital, Professional, Ancillary, or similar. This will likely only be listed once per page, if at all.
 
+If there are multiple correct answers for any of the above attributes, return only the answer corresponding to the SECOND of the two pages.
+If there are multiple correct answers for the second page, return them both in a comma-separated list. However, in most instances there will be only one.
+
+Return N/A for any answers not found.
+
 Only return the dictionary, with no other commentary or explanation. Ensure you abide by proper JSON formatting.
 """
 
@@ -95,7 +98,7 @@ Note that a 'lesser of'/'greater of' statement may not be directly written for a
 Read and analyze the page, then populate a JSON dictionary with the following fields (descriptions provided):
 'LESSER_OF_LANGUAGE_IND' : Write 'Y' if there is 'lesser of' language somewhere on the page that applies to the Service and Methodology. Write 'N' if not.
 'GREATER_OF_LANGUAGE_IND' : Write 'Y' if there is 'greater of' language somewhere on the page that applies to the Service and Methodology. Write 'N' if not.
-'FULL_METHODOLOGY' : If either of the above are 'Y', what is the reimbursement the lesser or greater of? Do not write the full sentence. Do not repeat the Methodology from above.
+'FULL_METHODOLOGY' : If either of the above are 'Y', what is the reimbursement the lesser or greater of? Do not write the full sentence - only include the first half of the lesser/greater of statement. For example, if the reimbursement is the lesser of X or Y, return X.
 'REIMBURSEMENT_RATE' : If the REIMBURSEMENT_METHODOLOGY found above is a percent of something, write just the numeric percent value here. If not, write N/A. Note that this is a separate value from that found in the Methodology above.
 'REIMBURSEMENT_FLAT_FEE' : If the REIMBURSEMENT_METHODOLOGY found above is a dollar value, write just the numeric value here. If not, write N/A. Note that this is a separate value from that found in the Methodology above.
 
@@ -134,8 +137,17 @@ Only return the dictionary, with no other commentary or explanation. Ensure you
 """
 
 def BOTTOM_UP_METHODOLOGY(d):
-    return f"""
+    return f"""Analyze the full methodology text listed below:
+{d['FULL_METHODOLOGY']}
 
+Your job is to identify what the concise methodology is based on the full methodology.
+
+Choose the closest option from the following:
+Allowed Amount, Fee Schedule, Medicare Fee Schedule, Medicare Allowed Amount, Medicaid Fee Schedule, Medicaid Allowed Amount, Billed Charges, Per X (where X is some criteria like Visit, Hour, Unit, Diem, etc)
+
+For 'Allowed Amount' and 'Fee Schedule', use the more specific answer (Medicare/Medicaid) if applicable.
+
+Return only the answer, without using complete sentences. If there is nothing even close to the above options in the full methodology text, simply return 'N/A'
 """
 
 

src/utils.py

@@ -39,39 +39,39 @@ def read_s3():
         files[filename] = context
     return files
 
-def read_input(mode=config.READ_MODE):
+def read_input(path=config.LOCAL_PATH, mode=config.READ_MODE):
     if mode == '_LOCAL_':
         files = {}
-        for file in os.listdir(config.LOCAL_PATH):
-            full_path = os.path.join(config.LOCAL_PATH, file)
+        for file in os.listdir(path):
+            full_path = os.path.join(path, file)
             file_text = read_local(full_path)
             files[file] = file_text
         return files
     elif mode == '_S3_':
         return read_s3()
 
-def consolidate_individual(folder):
+def consolidate_individual(input_folder='results', output_folder='output'):
     dfs = []
-    for filename in os.listdir(folder):
+    for filename in os.listdir(input_folder):
         if filename.endswith('.csv'):
-            filepath = os.path.join(folder, filename)
+            filepath = os.path.join(input_folder, filename)
             dfs.append(pd.read_csv(filepath))
     
     # Remove temp folder
-    if folder=='temp' and os.path.exists(folder):
-        shutil.rmtree(folder)
+    if input_folder=='temp' and os.path.exists(input_folder):
+        shutil.rmtree(input_folder)
     
     consolidated_df = pd.concat(dfs, ignore_index=True)
 
     # Write output
     version = 1
-    existing_files = [filename for filename in os.listdir('results') if filename.startswith(f'consolidated_results_{config.TODAY}')]
+    existing_files = [filename for filename in os.listdir(output_folder) if filename.startswith(f'consolidated_results_{config.TODAY}')]
     if existing_files:
         versions = [int(file.split('_v')[1].split('.')[0]) for file in existing_files if '_v' in file]
         if versions:
             version = max(versions) + 1
     filename = f'consolidated_results_{config.TODAY}_v{version}.csv'
-    consolidated_df.to_csv(os.path.join('results', filename))
+    consolidated_df.to_csv(os.path.join(output_folder, filename))