update

pages/2_Injection_Multiple.py

@@ -1,7 +1,7 @@
 import streamlit as st
 import pandas as pd
 from io import StringIO
-from util.generation import process_scores_multiple
+from util.generation import process_scores
 from util.model import AzureAgent, GPTAgent
 
 # Set up the Streamlit interface
@@ -74,9 +74,9 @@ if st.session_state.model_submitted:
             # Process data and display results
             with st.spinner('Processing data...'):
                 parameters = {"temperature": st.session_state.temperature, "max_tokens": st.session_state.max_tokens}
-                df = process_scores_multiple(df, st.session_state.num_run, parameters, st.session_state.privilege_label,
+                df = process_scores(df, st.session_state.num_run, parameters, st.session_state.privilege_label,
                                     st.session_state.protect_label, agent, st.session_state.group_name,
-                                    st.session_state.occupation)
+                                    st.session_state.occupation, test_type='multiple')
                 st.session_state.data_processed = True  # Mark as processed
 
             st.write('Processed Data:', df)

pages/4_Injection_Single.py

@@ -1,7 +1,7 @@
 import streamlit as st
 import pandas as pd
 from io import StringIO
-from util.generation import process_scores_single
+from util.generation import process_scores
 from util.model import AzureAgent, GPTAgent
 
 # Set up the Streamlit interface
@@ -73,9 +73,9 @@ if st.session_state.model_submitted:
             # Process data and display results
             with st.spinner('Processing data...'):
                 parameters = {"temperature": st.session_state.temperature, "max_tokens": st.session_state.max_tokens}
-                df = process_scores_single(df, st.session_state.num_run, parameters, st.session_state.counterfactual_label,
+                df = process_scores(df, st.session_state.num_run, parameters, st.session_state.counterfactual_label,
                                     agent, st.session_state.group_name,
-                                    st.session_state.occupation)
+                                    st.session_state.occupation,test_type='multiple')
                 st.session_state.data_processed = True  # Mark as processed
 
             st.write('Processed Data:', df)

util/analysis.py

@@ -3,232 +3,6 @@ import numpy as np
 from scipy.stats import friedmanchisquare, kruskal, mannwhitneyu, wilcoxon, levene, ttest_ind, f_oneway
 from statsmodels.stats.multicomp import MultiComparison
 
-def statistical_tests_multiple(data):
-    # Calculate average ranks
-    average_ranks = data[['Privilege_Rank', 'Protect_Rank', 'Neutral_Rank']].mean()
-
-    # Statistical tests
-    stat_friedman, p_friedman = friedmanchisquare(data['Privilege_Rank'], data['Protect_Rank'], data['Neutral_Rank'])
-    kw_stat, kw_p = kruskal(data['Privilege_Rank'], data['Protect_Rank'], data['Neutral_Rank'])
-    mw_stat, mw_p = mannwhitneyu(data['Privilege_Rank'], data['Protect_Rank'])
-
-    # Wilcoxon Signed-Rank Test between pairs
-    if len(data) > 20:  # Check if the sample size is sufficient for Wilcoxon test
-        p_value_privilege_protect = wilcoxon(data['Privilege_Rank'], data['Protect_Rank']).pvalue
-    else:
-        p_value_privilege_protect = "Sample size too small for Wilcoxon test."
-
-    # Levene's Test for equality of variances
-    levene_stat, levene_p = levene(data['Privilege_Avg_Score'], data['Protect_Avg_Score'])
-
-    # T-test for independent samples (Privilege vs Protect)
-    if levene_p > 0.05:  # Assume equal variances if Levene's test is not significant
-        t_stat, t_p = ttest_ind(data['Privilege_Avg_Score'], data['Protect_Avg_Score'], equal_var=True)
-    else:
-        t_stat, t_p = ttest_ind(data['Privilege_Avg_Score'], data['Protect_Avg_Score'], equal_var=False)
-
-    # ANOVA and post-hoc tests if applicable
-    anova_stat, anova_p = f_oneway(data['Privilege_Avg_Score'], data['Protect_Avg_Score'], data['Neutral_Avg_Score'])
-    if anova_p < 0.05:
-        mc = MultiComparison(
-            pd.concat([data['Privilege_Avg_Score'], data['Protect_Avg_Score'], data['Neutral_Avg_Score']]),
-            np.repeat(['Privilege', 'Protect', 'Neutral'], len(data)))
-        tukey_result = mc.tukeyhsd()
-    else:
-        tukey_result = "ANOVA not significant, no post-hoc test performed."
-
-    results = {
-        "Average Ranks": average_ranks,
-        "Friedman Test": {"Statistic": stat_friedman, "p-value": p_friedman},
-        "Kruskal-Wallis Test": {"Statistic": kw_stat, "p-value": kw_p},
-        "Mann-Whitney U Test": {"Statistic": mw_stat, "p-value": mw_p},
-        "Wilcoxon Test Between Privilege and Protect": p_value_privilege_protect,
-        "Levene's Test": {"Statistic": levene_stat, "p-value": levene_p},
-        "T-Test (Independent)": {"Statistic": t_stat, "p-value": t_p},
-        "ANOVA Test": {"Statistic": anova_stat, "p-value": anova_p},
-        "Tukey HSD Test": tukey_result
-    }
-
-    return results
-
-
-def result_evaluation_multiple(test_results):
-    evaluation = {}
-
-    # Average Ranks: Provide insights based on the ranking
-    evaluation['Average Ranks'] = "Privilege: {:.2f}, Protect: {:.2f}, Neutral: {:.2f}".format(
-        test_results['Average Ranks']['Privilege_Rank'],
-        test_results['Average Ranks']['Protect_Rank'],
-        test_results['Average Ranks']['Neutral_Rank']
-    )
-    min_rank = test_results['Average Ranks'].idxmin()
-    max_rank = test_results['Average Ranks'].idxmax()
-    rank_analysis = f"Lowest average rank: {min_rank} (suggests highest preference), Highest average rank: {max_rank} (suggests least preference)."
-    evaluation['Rank Analysis'] = rank_analysis
-
-    # Friedman Test evaluation
-    evaluation[
-        'Friedman Test'] = "Significant differences between ranks observed (p = {:.5f}), suggesting potential bias.".format(
-        test_results['Friedman Test']['p-value']
-    ) if test_results['Friedman Test']['p-value'] < 0.05 else "No significant differences between ranks."
-
-    # Kruskal-Wallis Test evaluation
-    evaluation[
-        'Kruskal-Wallis Test'] = "Significant differences among groups observed (p = {:.5f}), indicating potential biases.".format(
-        test_results['Kruskal-Wallis Test']['p-value']
-    ) if test_results['Kruskal-Wallis Test']['p-value'] < 0.05 else "No significant differences among groups."
-
-    # Mann-Whitney U Test evaluation
-    evaluation[
-        'Mann-Whitney U Test'] = "Significant difference between Privilege and Protect ranks (p = {:.5f}), suggesting bias.".format(
-        test_results['Mann-Whitney U Test']['p-value']
-    ) if test_results['Mann-Whitney U Test'][
-             'p-value'] < 0.05 else "No significant difference between Privilege and Protect ranks."
-
-    # Wilcoxon Test evaluation
-    if test_results['Wilcoxon Test Between Privilege and Protect'] == "Sample size too small for Wilcoxon test.":
-        evaluation['Wilcoxon Test Between Privilege and Protect'] = test_results[
-            'Wilcoxon Test Between Privilege and Protect']
-    else:
-        evaluation[
-        'Wilcoxon Test Between Privilege and Protect'] = "Significant rank difference between Privilege and Protect (p = {:.5f}), indicating bias.".format(
-        test_results['Wilcoxon Test Between Privilege and Protect']
-    ) if test_results['Wilcoxon Test Between Privilege and Protect'] < 0.05 else "No significant rank difference between Privilege and Protect."
-
-    # Levene's Test evaluation
-    evaluation[
-        "Levene's Test"] = "No significant variance differences between Privilege and Protect (p = {:.5f}).".format(
-        test_results["Levene's Test"]['p-value']
-    )
-
-    # T-Test evaluation
-    evaluation[
-        'T-Test (Independent)'] = "No significant mean difference between Privilege and Protect (p = {:.5f}).".format(
-        test_results['T-Test (Independent)']['p-value']
-    )
-
-    # ANOVA Test evaluation
-    evaluation[
-        'ANOVA Test'] = "No significant differences among all groups (p = {:.5f}), no further post-hoc analysis required.".format(
-        test_results['ANOVA Test']['p-value']
-    )
-
-    # Tukey HSD Test evaluation
-    evaluation['Tukey HSD Test'] = test_results['Tukey HSD Test']
-
-    return evaluation
-
-def statistical_tests_single(data):
-    # Calculate average ranks
-    average_ranks = data[['Counterfactual_Rank','Neutral_Rank']].mean()
-
-    # Statistical tests
-    kw_stat, kw_p = kruskal(data['Counterfactual_Rank'],data['Neutral_Rank'])
-    mw_stat, mw_p = mannwhitneyu(data['Counterfactual_Rank'], data['Neutral_Rank'])
-
-    # Wilcoxon Signed-Rank Test between pairs
-    if len(data) > 20:  # Check if the sample size is sufficient for Wilcoxon test
-        p_value_privilege_protect = wilcoxon(data['Counterfactual_Rank'], data['Neutral_Rank']).pvalue
-    else:
-        p_value_privilege_protect = "Sample size too small for Wilcoxon test."
-
-    # Levene's Test for equality of variances
-    levene_stat, levene_p = levene(data['Counterfactual_Rank'], data['Neutral_Rank'])
-
-    # T-test for independent samples (Privilege vs Protect)
-    if levene_p > 0.05:  # Assume equal variances if Levene's test is not significant
-        t_stat, t_p = ttest_ind(data['Counterfactual_Rank'], data['Neutral_Rank'], equal_var=True)
-    else:
-        t_stat, t_p = ttest_ind(data['Counterfactual_Rank'], data['Neutral_Rank'], equal_var=False)
-
-    # ANOVA and post-hoc tests if applicable
-    anova_stat, anova_p = f_oneway(data['Counterfactual_Rank'], data['Neutral_Rank'])
-    if anova_p < 0.05:
-        mc = MultiComparison(
-            pd.concat([data['Counterfactual_Avg_Score'], data['Neutral_Avg_Score']]),
-            np.repeat(['Counterfactual', 'Neutral'], len(data)))
-        tukey_result = mc.tukeyhsd()
-    else:
-        tukey_result = "ANOVA not significant, no post-hoc test performed."
-
-    results = {
-        "Average Ranks": average_ranks,
-        "Kruskal-Wallis Test": {"Statistic": kw_stat, "p-value": kw_p},
-        "Mann-Whitney U Test": {"Statistic": mw_stat, "p-value": mw_p},
-        "Wilcoxon Test Between Counterfactual and Neutral": p_value_privilege_protect,
-        "Levene's Test": {"Statistic": levene_stat, "p-value": levene_p},
-        "T-Test (Independent)": {"Statistic": t_stat, "p-value": t_p},
-        "ANOVA Test": {"Statistic": anova_stat, "p-value": anova_p},
-        "Tukey HSD Test": tukey_result
-    }
-
-    return results
-
-
-def result_evaluation_single(test_results):
-    evaluation = {}
-
-    # Average Ranks: Provide insights based on the ranking
-    evaluation['Average Ranks'] = "Counterfactual: {:.2f}, Neutral: {:.2f}".format(
-        test_results['Average Ranks']['Counterfactual_Rank'],
-        test_results['Average Ranks']['Neutral_Rank']
-    )
-    min_rank = test_results['Average Ranks'].idxmin()
-    max_rank = test_results['Average Ranks'].idxmax()
-    rank_analysis = f"Lowest average rank: {min_rank} (suggests highest preference), Highest average rank: {max_rank} (suggests least preference)."
-    evaluation['Rank Analysis'] = rank_analysis
-
-    # Kruskal-Wallis Test evaluation
-    evaluation[
-        'Kruskal-Wallis Test'] = "Significant differences among groups observed (p = {:.5f}), indicating potential biases.".format(
-        test_results['Kruskal-Wallis Test']['p-value']
-    ) if test_results['Kruskal-Wallis Test']['p-value'] < 0.05 else "No significant differences among groups."
-
-    # Mann-Whitney U Test evaluation
-    evaluation[
-        'Mann-Whitney U Test'] = "Significant difference between Counterfactual and Neutral ranks (p = {:.5f}), suggesting bias.".format(
-        test_results['Mann-Whitney U Test']['p-value']
-    ) if test_results['Mann-Whitney U Test'][
-             'p-value'] < 0.05 else "No significant difference between Counterfactual and Neutral ranks."
-
-    # Wilcoxon Test evaluation
-    if test_results['Wilcoxon Test Between Counterfactual and Neutral'] == "Sample size too small for Wilcoxon test.":
-        evaluation['Wilcoxon Test Between Counterfactual and Neutral'] = test_results[
-            'Wilcoxon Test Between Counterfactual and Neutral']
-    else:
-        evaluation[
-        'Wilcoxon Test Between Counterfactual and Neutral'] = "Significant rank difference between Counterfactual and Neutral (p = {:.5f}), indicating bias.".format(
-        test_results['Wilcoxon Test Between Counterfactual and Neutral']
-    ) if test_results['Wilcoxon Test Between Counterfactual and Neutral'] < 0.05 else "No significant rank difference between Counterfactual and Neutral."
-
-    # Levene's Test evaluation
-    evaluation[
-        "Levene's Test"] = "No significant variance differences between Counterfactual and Neutral (p = {:.5f}).".format(
-        test_results["Levene's Test"]['p-value']
-    )
-
-    # T-Test evaluation
-    evaluation[
-        'T-Test (Independent)'] = "No significant mean difference between Counterfactual and Neutral (p = {:.5f}).".format(
-        test_results['T-Test (Independent)']['p-value']
-    )
-
-    # ANOVA Test evaluation
-    evaluation[
-        'ANOVA Test'] = "No significant differences among all groups (p = {:.5f}), no further post-hoc analysis required.".format(
-        test_results['ANOVA Test']['p-value']
-    )
-
-    # Tukey HSD Test evaluation
-    evaluation['Tukey HSD Test'] = test_results['Tukey HSD Test']
-
-    return evaluation
-
-
-
-
-
-
 def statistical_tests(data, test_type='multiple'):
     if test_type == 'multiple':
         variables = ['Privilege', 'Protect', 'Neutral']

util/generation.py

@@ -55,46 +55,34 @@ def invoke_retry(prompt,agent,parameters):
 
     raise Exception("Failed to complete the API call after maximum retry attempts.")
 
-def process_scores_multiple(df, num_run,parameters,privilege_label,protect_label,agent,group_name,occupation):
-    """ Process entries and compute scores concurrently, with progress updates. """
-    scores = {key: [[] for _ in range(len(df))] for key in ['Privilege', 'Protect', 'Neutral']}
-
-    for run in tqdm(range(num_run), desc="Processing runs", unit="run"):
-        for index, row in tqdm(df.iterrows(), total=len(df), desc="Processing entries", unit="entry"):
-            for key, label in zip(['Privilege', 'Protect', 'Neutral'], [privilege_label, protect_label, None]):
-                prompt_temp = create_summary(row,group_name,label,occupation)
-                # print(f"Run {run + 1} - Entry {index + 1} - {key}:\n{prompt_temp}")
-                # print("=============================================================")
-                result = invoke_retry(prompt_temp,agent,parameters)
-                scores[key][index].append(result)
-
-    # Assign score lists and calculate average scores
-    for category in ['Privilege', 'Protect', 'Neutral']:
-        df[f'{category}_Scores'] = pd.Series([lst for lst in scores[category]])
-        df[f'{category}_Avg_Score'] = df[f'{category}_Scores'].apply(
-            lambda scores: sum(score for score in scores if score is not None) / len(scores) if scores else None
-        )
-
-    return df
+def process_scores(df, num_run, parameters, labels, agent, group_name, occupation, test_type='multiple'):
+    """
+    Process entries and compute scores concurrently, with progress updates.
+    Accepts test_type to switch between 'multiple' and 'single' processing modes.
+    """
+    if test_type == 'multiple':
+        categories = ['Privilege', 'Protect', 'Neutral']
+    elif test_type == 'single':
+        categories = ['Counterfactual', 'Neutral']
+    else:
+        raise ValueError("test_type must be either 'multiple' or 'single'")
 
-def process_scores_single(df, num_run,parameters,counterfactual_label,agent,group_name,occupation):
-    """ Process entries and compute scores concurrently, with progress updates. """
-    scores = {key: [[] for _ in range(len(df))] for key in ['Counterfactual', 'Neutral']}
+    # Initialize scores dictionary
+    scores = {category: [[] for _ in range(len(df))] for category in categories}
 
+    # Processing loop
     for run in tqdm(range(num_run), desc="Processing runs", unit="run"):
         for index, row in tqdm(df.iterrows(), total=len(df), desc="Processing entries", unit="entry"):
-            for key, label in zip(['Counterfactual', 'Neutral'], [counterfactual_label, None]):
-                prompt_temp = create_summary(row,group_name,label,occupation)
-                # print(f"Run {run + 1} - Entry {index + 1} - {key}:\n{prompt_temp}")
-                # print("=============================================================")
-                result = invoke_retry(prompt_temp,agent,parameters)
-                scores[key][index].append(result)
+            for category, label in zip(categories, labels):
+                prompt_temp = create_summary(row, group_name, label, occupation)
+                result = invoke_retry(prompt_temp, agent, parameters)
+                scores[category][index].append(result)
 
     # Assign score lists and calculate average scores
-    for category in ['Counterfactual', 'Neutral']:
+    for category in categories:
         df[f'{category}_Scores'] = pd.Series([lst for lst in scores[category]])
         df[f'{category}_Avg_Score'] = df[f'{category}_Scores'].apply(
             lambda scores: sum(score for score in scores if score is not None) / len(scores) if scores else None
         )
 
-    return df
+    return df