Spaces:
Sleeping
Sleeping
File size: 9,990 Bytes
f533950 6655d5a f533950 6655d5a d7de222 6655d5a d7de222 6655d5a d7de222 6655d5a f533950 6655d5a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 |
import gradio as gr
import json
import logging
from visualization.bow_visualizer import process_and_visualize_analysis
from processors.topic_modeling import compare_topics
from processors.ngram_analysis import compare_ngrams
from processors.bow_analysis import compare_bow
from processors.text_classifiers import classify_formality, classify_sentiment, classify_complexity, compare_classifications
# Add import for bias detection
from processors.bias_detection import compare_bias
# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger('analysis_handler')
def process_analysis_request(dataset, selected_analysis, parameters):
"""
Process the analysis request based on the selected options.
Args:
dataset (dict): The input dataset
selected_analysis (str): The selected analysis type
parameters (dict): Additional parameters for the analysis
Returns:
tuple: A tuple containing (analysis_results, visualization_data)
"""
logger.info(f"Processing analysis request: {selected_analysis}")
if not dataset or "entries" not in dataset or not dataset["entries"]:
logger.warning("No valid dataset provided for analysis")
return {}, None
# Initialize the results structure
results = {"analyses": {}}
# Get the prompt text from the first entry
prompt_text = dataset["entries"][0].get("prompt", "")
if not prompt_text:
logger.warning("No prompt found in dataset")
return {"error": "No prompt found in dataset"}, None
# Initialize the analysis container for this prompt
results["analyses"][prompt_text] = {}
# Get model names and responses
model1_name = dataset["entries"][0].get("model", "Model 1")
model2_name = dataset["entries"][1].get("model", "Model 2")
model1_response = dataset["entries"][0].get("response", "")
model2_response = dataset["entries"][1].get("response", "")
logger.info(f"Comparing responses from {model1_name} and {model2_name}")
try:
# Process based on the selected analysis type
if selected_analysis == "Bag of Words":
# Get the top_n parameter and ensure it's an integer
top_n = parameters.get("bow_top", 25)
if isinstance(top_n, str):
top_n = int(top_n)
logger.info(f"Running Bag of Words analysis with top_n={top_n}")
# Perform Bag of Words analysis using the processor
bow_results = compare_bow(
[model1_response, model2_response],
[model1_name, model2_name],
top_n=top_n
)
results["analyses"][prompt_text]["bag_of_words"] = bow_results
elif selected_analysis == "N-gram Analysis":
# Perform N-gram analysis
ngram_size = parameters.get("ngram_n", 2)
if isinstance(ngram_size, str):
ngram_size = int(ngram_size)
top_n = parameters.get("ngram_top", 15)
if isinstance(top_n, str):
top_n = int(top_n)
logger.info(f"Running N-gram analysis with n={ngram_size}, top_n={top_n}")
# Use the processor from the dedicated ngram_analysis module
from processors.ngram_analysis import compare_ngrams as ngram_processor
ngram_results = ngram_processor(
[model1_response, model2_response],
[model1_name, model2_name],
n=ngram_size,
top_n=top_n
)
results["analyses"][prompt_text]["ngram_analysis"] = ngram_results
elif selected_analysis == "Topic Modeling":
# Perform topic modeling analysis
topic_count = parameters.get("topic_count", 3)
if isinstance(topic_count, str):
topic_count = int(topic_count)
logger.info(f"Running Topic Modeling analysis with n_topics={topic_count}")
try:
# Import the improved topic modeling module
try:
# First try to import from improved module if available
from improved_topic_modeling import compare_topics as improved_compare_topics
logger.info("Using improved topic modeling implementation")
topic_results = improved_compare_topics(
texts_set_1=[model1_response],
texts_set_2=[model2_response],
n_topics=topic_count,
model_names=[model1_name, model2_name])
except ImportError:
# Fall back to original implementation
logger.info("Using original topic modeling implementation")
from processors.topic_modeling import compare_topics
topic_results = compare_topics(
texts_set_1=[model1_response],
texts_set_2=[model2_response],
n_topics=topic_count,
model_names=[model1_name, model2_name])
results["analyses"][prompt_text]["topic_modeling"] = topic_results
# Ensure the topic modeling results contain the necessary fields
if "topics" not in topic_results or not topic_results["topics"]:
logger.warning("No topics found in topic modeling results")
topic_results["message"] = "No significant topics were discovered in the text. Try a different analysis method or adjust parameters."
if "model_topics" not in topic_results or not topic_results["model_topics"]:
logger.warning("No model topics found in topic modeling results")
if "message" not in topic_results:
topic_results["message"] = "Could not calculate topic distributions for the models."
except Exception as e:
import traceback
error_msg = f"Topic modeling error: {str(e)}\n{traceback.format_exc()}"
logger.error(error_msg)
results["analyses"][prompt_text]["topic_modeling"] = {
"models": [model1_name, model2_name],
"error": str(e),
"message": "Topic modeling failed. Please try with longer text or different parameters.",
"stack_trace": traceback.format_exc()
}
elif selected_analysis == "Classifier":
# Perform classifier analysis
logger.info("Running Classifier analysis")
results["analyses"][prompt_text]["classifier"] = {
"models": [model1_name, model2_name],
"classifications": {
model1_name: {
"formality": classify_formality(model1_response),
"sentiment": classify_sentiment(model1_response),
"complexity": classify_complexity(model1_response)
},
model2_name: {
"formality": classify_formality(model2_response),
"sentiment": classify_sentiment(model2_response),
"complexity": classify_complexity(model2_response)
}
},
"differences": compare_classifications(model1_response, model2_response)
}
elif selected_analysis == "Bias Detection":
# Perform bias detection analysis
logger.info("Running Bias Detection analysis")
try:
# Perform bias detection analysis
logger.info(f"Starting bias detection for {model1_name} and {model2_name}")
logger.info(f"Text lengths - Text1: {len(model1_response)}, Text2: {len(model2_response)}")
bias_results = compare_bias(
model1_response,
model2_response,
model_names=[model1_name, model2_name]
)
logger.info(f"Bias detection complete. Result has keys: {bias_results.keys() if bias_results else 'None'}")
results["analyses"][prompt_text]["bias_detection"] = bias_results
except Exception as e:
import traceback
error_msg = f"Bias detection error: {str(e)}\n{traceback.format_exc()}"
logger.error(error_msg)
results["analyses"][prompt_text]["bias_detection"] = {
"models": [model1_name, model2_name],
"error": str(e),
"message": "Bias detection failed. Try with different parameters.",
"stack_trace": traceback.format_exc()
}
else:
# Unknown analysis type
logger.warning(f"Unknown analysis type: {selected_analysis}")
results["analyses"][prompt_text]["message"] = "Please select a valid analysis type."
except Exception as e:
import traceback
error_msg = f"Error processing analysis request: {str(e)}\n{traceback.format_exc()}"
logger.error(error_msg)
results = {
"error": str(e),
"stack_trace": traceback.format_exc(),
"analyses": {
prompt_text: {
"message": f"Analysis failed: {str(e)}"
}
}
}
# Return both the analysis results and a placeholder for visualization data
return results, None |