Spaces:
Sleeping
Sleeping
File size: 5,900 Bytes
9908b22 | 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 | """
Enhanced multi-agent system with dynamic marks configuration
"""
import json
import time
from typing import Dict, Any, Tuple
from agents import GeneratorAgent, VerifierAgent, FormatterAgent, SearchAgent
from prompts import get_generator_prompt, VERIFIER_PROMPT, FORMATTER_PROMPT
class MultiAgentSystem:
"""Enhanced orchestrator with dynamic marks support"""
def __init__(self):
self.generator = GeneratorAgent()
self.verifier = VerifierAgent()
self.formatter = FormatterAgent()
self.searcher = SearchAgent()
self.progress = {
"stage": "ready",
"message": "System ready",
"percentage": 0
}
def update_progress(self, stage: str, message: str, percentage: int):
"""Update generation progress"""
self.progress = {
"stage": stage,
"message": message,
"percentage": percentage
}
print(f"Progress: {stage} - {message} ({percentage}%)")
def extract_syllabus_from_pdf(self, pdf_file) -> str:
"""Extract text from uploaded syllabus PDF"""
try:
import PyPDF2
pdf_reader = PyPDF2.PdfReader(pdf_file)
text = ""
for page in pdf_reader.pages:
text += page.extract_text()
return text[:3000]
except Exception as e:
print(f"PDF extraction error: {e}")
return "Syllabus text extraction failed. Using default template."
def extract_reference_from_pdf(self, pdf_file) -> str:
"""Extract text from reference question paper PDF"""
if pdf_file is None:
return "No reference question paper provided."
try:
import PyPDF2
pdf_reader = PyPDF2.PdfReader(pdf_file)
text = ""
for page in pdf_reader.pages[:3]:
text += page.extract_text()
return text[:2000]
except Exception as e:
print(f"Reference PDF extraction error: {e}")
return "Reference paper extraction failed."
def generate_exam_package(self,
subject: str,
stream: str,
part_a_count: int,
part_b_count: int,
part_c_count: int,
part_a_marks: int = 2,
part_b_marks: int = 13,
part_c_marks: int = 14,
syllabus_file = None,
reference_file = None) -> Tuple[Dict[str, Any], str]:
"""Enhanced main method with dynamic marks"""
try:
# Stage 1: Data Preparation
self.update_progress("preparation", "Extracting syllabus and reference data", 10)
syllabus_text = self.extract_syllabus_from_pdf(syllabus_file)
reference_text = self.extract_reference_from_pdf(reference_file)
# Stage 2: Realtime Search
self.update_progress("search", "Fetching recent developments", 20)
realtime_updates = self.searcher.get_realtime_updates(subject)
# Stage 3: Generation with dynamic marks
self.update_progress("generation", "Generating question paper structure", 40)
generator_prompt = get_generator_prompt(
subject=subject,
stream=stream,
syllabus_text=syllabus_text,
reference_text=reference_text,
realtime_updates=realtime_updates,
part_a_count=part_a_count,
part_b_count=part_b_count,
part_c_count=part_c_count,
part_a_marks=part_a_marks,
part_b_marks=part_b_marks,
part_c_marks=part_c_marks
)
generated_content = self.generator.generate_question_paper(generator_prompt)
if "error" in generated_content:
return {}, "Generation failed: " + generated_content["error"]
# Stage 4: Verification with dynamic marks
self.update_progress("verification", "Verifying quality and standards", 60)
verifier_prompt = VERIFIER_PROMPT.format(
generated_content=json.dumps(generated_content, indent=2),
bloom_mix="60% R/U, 40% A/An/Ev" if stream == "CSE" else "50% R/U, 50% A/An/Ev",
part_a_count=part_a_count,
part_b_count=part_b_count,
part_c_count=part_c_count,
part_a_marks=part_a_marks,
part_b_marks=part_b_marks,
part_c_marks=part_c_marks,
tag_requirements="Company tags" if stream == "CSE" else "GATE tags"
)
verification_result = self.verifier.verify_content(verifier_prompt)
# Stage 5: Formatting
self.update_progress("formatting", "Creating final outputs and answers", 80)
formatter_prompt = FORMATTER_PROMPT.format(
original_content=json.dumps(generated_content, indent=2),
corrections=json.dumps(verification_result, indent=2)
)
final_output = self.formatter.format_final_output(formatter_prompt)
# Stage 6: Completion
self.update_progress("completion", "Package generation complete", 100)
return final_output, "Success"
except Exception as e:
error_msg = f"System error: {str(e)}"
print(error_msg)
return {}, error_msg
def get_progress(self) -> Dict[str, Any]:
"""Get current progress"""
return self.progress |