SPJIMR_FlipClassroom_RCopilot_ResearchInternship / subjective_test_evaluation.py
omkar-surve126's picture
Upload 2 files
d45297d verified
raw
history blame
9.8 kB
import openai
from pymongo import MongoClient
from datetime import datetime
import os
from dotenv import load_dotenv
import re
import streamlit as st
from bson import ObjectId
load_dotenv()
MONGO_URI = os.getenv('MONGO_URI')
OPENAI_API_KEY = os.getenv('OPENAI_KEY')
client = MongoClient(MONGO_URI)
db = client['novascholar_db']
rubrics_collection = db['rubrics']
resources_collection = db['resources']
subjective_tests_collection = db['subjective_tests']
subjective_test_analysis_collection = db['subjective_test_analysis']
openai.api_key = OPENAI_API_KEY
def evaluate_subjective_answers(test_id, student_id, course_id):
"""Evaluate subjective test answers using OpenAI."""
try:
# Get test and submission details
test_doc = subjective_tests_collection.find_one({
"_id": ObjectId(test_id),
"course_id": course_id
})
if not test_doc:
return {
"content_analysis": "Error: Test not found",
"analyzed_at": datetime.utcnow(),
"correctness_score": 0
}
submission = next(
(sub for sub in test_doc.get('submissions', []) if sub['student_id'] == student_id),
None
)
if not submission:
return {
"content_analysis": "Error: Submission not found",
"analyzed_at": datetime.utcnow(),
"correctness_score": 0
}
# Rest of the evaluation logic remains the same
questions = test_doc.get('questions', [])
student_answers = submission.get('answers', [])
if not questions or not student_answers:
return {
"content_analysis": "Error: No questions or answers found",
"analyzed_at": datetime.utcnow(),
"correctness_score": 0
}
# Retrieve rubrics for the session
rubric_doc = rubrics_collection.find_one({
"session_id": test_doc['session_id'],
"course_id": course_id
})
if not rubric_doc:
return {
"content_analysis": "Error: Rubric not found",
"analyzed_at": datetime.utcnow(),
"correctness_score": 0
}
rubric = rubric_doc.get('rubric', {})
# Retrieve pre-class materials
pre_class_materials = resources_collection.find({
"session_id": test_doc['session_id'],
"course_id": course_id
})
pre_class_content = "\n".join([material.get('text_content', '') for material in pre_class_materials])
# Analyze each question
all_analyses = []
total_score = 0
for i, (question, answer) in enumerate(zip(questions, student_answers), 1):
analysis_content = f"Question {i}: {question['question']}\nAnswer: {answer}\n\nRubric: {rubric}\n\nPre-class Materials: {pre_class_content}\n\n"
prompt_template = f"""As an educational assessor, evaluate this student's answer based on the provided rubric criteria and pre-class materials. Follow these assessment guidelines:
1. Evaluation Process:
- Use each rubric criterion (scored 1-4) for internal assessment
- Compare response with pre-class materials
- Check alignment with all rubric requirements
- Calculate final score: sum of criteria scores converted to 10-point scale
Pre-class Materials:
{pre_class_content}
Rubric Criteria:
{rubric}
Question and Answer:
{analysis_content}
Provide your assessment in the following format:
**Score and Evidence**
- Score: [X]/10
- Evidence for deduction: [One-line reference to most significant gap or inaccuracy]
**Key Areas for Improvement**
- [Concise improvement point 1]
- [Concise improvement point 2]
- [Concise improvement point 3]
"""
response = openai.Completion.create(
model="text-davinci-003",
prompt=prompt_template,
max_tokens=500,
temperature=0.7
)
individual_analysis = response.choices[0].text.strip()
try:
score_match = re.search(r'Score: (\d+)', individual_analysis)
question_score = int(score_match.group(1)) if score_match else 0
total_score += question_score
except:
question_score = 0
formatted_analysis = f"\n\n## Question {i} Analysis\n\n{individual_analysis}"
all_analyses.append(formatted_analysis)
average_score = round(total_score / len(questions)) if questions else 0
combined_analysis = "\n".join(all_analyses)
return {
"content_analysis": combined_analysis,
"analyzed_at": datetime.utcnow(),
"correctness_score": average_score
}
except Exception as e:
return {
"content_analysis": f"Error evaluating answers: {str(e)}",
"analyzed_at": datetime.utcnow(),
"correctness_score": 0
}
def display_evaluation_to_faculty(session_id, student_id, course_id):
"""Display submitted tests with improved error handling and debugging"""
st.subheader("Evaluate Subjective Tests")
try:
# Convert all IDs to strings for consistent comparison
session_id = str(session_id)
student_id = str(student_id)
course_id = str(course_id)
print(f"Searching for tests with session_id: {session_id}, student_id: {student_id}, course_id: {course_id}")
# Query for tests
query = {
"session_id": session_id,
"course_id": course_id,
"submissions": {
"$elemMatch": {
"student_id": student_id
}
}
}
# Log the query for debugging
print(f"MongoDB Query: {query}")
# Fetch tests
tests = list(subjective_tests_collection.find(query))
print(f"Found {len(tests)} tests matching query")
if not tests:
# Check if any tests exist for this session
all_session_tests = list(subjective_tests_collection.find({
"session_id": session_id,
"course_id": course_id
}))
if all_session_tests:
print(f"Found {len(all_session_tests)} tests for this session, but no submissions from student {student_id}")
st.warning("No submitted tests found for this student, but tests exist for this session.")
else:
print("No tests found for this session at all")
st.info("No tests have been created for this session yet.")
return
# Display tests and handle evaluation
for test in tests:
with st.expander(f"Test: {test.get('title', 'Untitled Test')}", expanded=True):
# Find student submission
submission = next(
(sub for sub in test.get('submissions', [])
if sub['student_id'] == student_id),
None
)
if submission:
st.write("### Student's Answers")
for i, (question, answer) in enumerate(zip(test['questions'], submission['answers'])):
st.markdown(f"**Q{i+1}:** {question['question']}")
st.markdown(f"**A{i+1}:** {answer}")
st.markdown("---")
# Generate/display analysis
if st.button(f"Generate Analysis for {test.get('title')}"):
with st.spinner("Analyzing responses..."):
analysis = evaluate_subjective_answers(
str(test['_id']),
student_id,
course_id
)
if analysis:
st.markdown("### Analysis")
st.markdown(analysis['content_analysis'])
st.metric("Score", f"{analysis['correctness_score']}/10")
else:
st.error("Submission data not found for this student")
except Exception as e:
st.error("An error occurred while loading the tests")
with st.expander("Error Details"):
st.write(f"Error: {str(e)}")
st.write(f"Session ID: {session_id}")
st.write(f"Student ID: {student_id}")
st.write(f"Course ID: {course_id}")
def check_test_submission(session_id, student_id, course_id):
"""Utility function to check test submission status"""
try:
query = {
"session_id": str(session_id),
"course_id": str(course_id),
"submissions.student_id": str(student_id)
}
test = subjective_tests_collection.find_one(query)
return bool(test)
except Exception as e:
print(f"Error checking submission: {e}")
return False