Create multi_agent.py

multi_agent.py

@@ -0,0 +1,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