Update app.py

app.py

@@ -6,7 +6,7 @@ import json
 import re
 import google.generativeai as genai
 from openai import OpenAI
-from typing import List, Dict, Tuple
+from typing import List, Dict, Optional
 
 class CognitiveArchitecture:
     def __init__(self):
@@ -18,300 +18,236 @@ class CognitiveArchitecture:
         }
         self.validate_keys()
         
-        # Initialize all AI models
         genai.configure(api_key=self.api_keys["GEMINI"])
-        self.gemini_model = genai.GenerativeModel(
-            "gemini-2.0-pro-exp-02-05",
-            generation_config={"temperature": 0.5, "max_output_tokens": 8192}
-        )
+        self.gemini_model = genai.GenerativeModel("gemini-2.0-pro-exp-02-05")
         
         self.gpt4o_client = OpenAI(
             base_url="https://models.inference.ai.azure.com",
             api_key=self.api_keys["AZURE"]
         )
 
-        self.models = {
-            "DeepSeek": "deepseek/deepseek-chat:free",
-            "Qwen": "qwen/qwen-vl-plus:free",
-            "Llama": "meta-llama/llama-3.3-70b-instruct:free",
-            "Mistral": "mistral-large-latest",
-            "GPT4o": "gpt-4o"
+        self.model_config = {
+            "decomposition": ["gemini-2.0-pro-exp-02-05", "mistral-large-latest"],
+            "analysis": ["gpt-4o"],
+            "critique": ["gpt-4o", "meta-llama/llama-3.3-70b-instruct:free"]
         }
 
         self.headers = {
-            "OpenRouter": {
-                "Authorization": f"Bearer {self.api_keys['OPENROUTER']}",
-                "Content-Type": "application/json"
-            },
-            "Mistral": {
-                "Authorization": f"Bearer {self.api_keys['MISTRAL']}",
-                "Content-Type": "application/json",
-                "Accept": "application/json"
-            }
+            "mistral": {"Authorization": f"Bearer {self.api_keys['MISTRAL']}"},
+            "openrouter": {"Authorization": f"Bearer {self.api_keys['OPENROUTER']}"}
         }
-        
-        self.memory = []
-        self.thinking_steps = []
 
     def validate_keys(self):
-        for key, value in self.api_keys.items():
-            if not value:
-                raise ValueError(f"Missing API key: {key}")
+        for service, key in self.api_keys.items():
+            if not key:
+                raise ValueError(f"Missing API key: {service}")
+
+    def safe_json_parse(self, text: str) -> dict:
+        """Robust JSON parsing with multiple fallbacks"""
+        try:
+            return json.loads(text)
+        except json.JSONDecodeError:
+            try:
+                # Try to extract JSON from text
+                json_str = re.search(r'\{.*\}', text, re.DOTALL)
+                if json_str:
+                    return json.loads(json_str.group())
+                # Fallback to key-value parsing
+                return self.parse_kv_text(text)
+            except Exception as e:
+                return {"error": f"JSON parsing failed: {str(e)}", "raw_response": text}
+
+    def parse_kv_text(self, text: str) -> dict:
+        """Fallback key-value parser"""
+        parsed = {}
+        current_key = None
+        lines = text.split('\n')
+        
+        for line in lines:
+            if ':' in line:
+                key, value = line.split(':', 1)
+                parsed[key.strip()] = value.strip()
+                current_key = key.strip()
+            elif current_key:
+                parsed[current_key] += " " + line.strip()
+        return parsed
 
-    def call_model(self, model_role: str, prompt: str, context: List[Dict] = None) -> str:
-        """Universal model router with error handling"""
+    def call_model(self, model: str, prompt: str) -> dict:
+        """Robust model caller with error handling"""
         try:
-            if model_role == "Gemini":
+            if "gemini" in model:
                 response = self.gemini_model.generate_content(prompt)
-                return response.text
+                return {"success": True, "content": response.text}
             
-            if model_role == "Mistral":
-                return self._call_mistral(prompt, context)
+            if "mistral" in model:
+                return self.call_mistral(model, prompt)
             
-            if model_role == "GPT4o":
-                return self._call_gpt4o(prompt, context)
+            if "gpt-4o" in model:
+                return self.call_azure_gpt4o(prompt)
             
-            # Handle OpenRouter models
-            payload = {
-                "model": self.models[model_role],
-                "messages": context if context else [{"role": "user", "content": prompt}],
-                "temperature": 0.7,
-                "max_tokens": 3096,
-                "top_p": 0.9
-            }
+            return self.call_openrouter(model, prompt)
             
-            response = requests.post(
-                "https://openrouter.ai/api/v1/chat/completions",
-                headers=self.headers["OpenRouter"],
-                json=payload,
-                timeout=30
-            )
-            
-            if response.status_code == 200:
-                return response.json()['choices'][0]['message']['content']
-            return f"API Error {response.status_code}: {response.text}"
-        
         except Exception as e:
-            print(f"Model Error ({model_role}): {str(e)}")
-            return ""
+            return {"success": False, "error": str(e)}
 
-    def _call_mistral(self, prompt: str, context: List[Dict] = None) -> str:
-        """Direct Mistral API call"""
+    def call_mistral(self, model: str, prompt: str) -> dict:
+        """Mistral API caller"""
         try:
-            payload = {
-                "model": self.models["Mistral"],
-                "messages": context if context else [{"role": "user", "content": prompt}],
-                "temperature": 0.7,
-                "max_tokens": 4096,
-                "top_p": 0.9
-            }
-            
             response = requests.post(
                 "https://api.mistral.ai/v1/chat/completions",
-                headers=self.headers["Mistral"],
-                json=payload,
+                headers=self.headers["mistral"],
+                json={
+                    "model": model,
+                    "messages": [{"role": "user", "content": prompt}],
+                    "temperature": 0.7,
+                    "max_tokens": 3000
+                },
                 timeout=30
             )
-            
-            return response.json()['choices'][0]['message']['content'] if response.status_code == 200 else ""
-
+            if response.status_code == 200:
+                return {"success": True, "content": response.json()['choices'][0]['message']['content']}
+            return {"success": False, "error": f"API Error {response.status_code}"}
         except Exception as e:
-            print(f"Mistral API Error: {str(e)}")
-            return ""
+            return {"success": False, "error": str(e)}
 
-    def _call_gpt4o(self, prompt: str, context: List[Dict] = None) -> str:
-        """Azure Inference API for GPT-4o"""
+    def call_azure_gpt4o(self, prompt: str) -> dict:
+        """Azure GPT-4o caller"""
         try:
-            messages = context if context else [
-                {"role": "system", "content": "You are an expert analyst."},
-                {"role": "user", "content": prompt}
-            ]
-            
             response = self.gpt4o_client.chat.completions.create(
-                model=self.models["GPT4o"],
-                messages=messages,
-                temperature=1.0,
-                top_p=1.0,
-                max_tokens=1000
+                model="gpt-4o",
+                messages=[{"role": "user", "content": prompt}],
+                temperature=0.7,
+                max_tokens=2000
             )
-            
-            return response.choices[0].message.content
-        
+            return {"success": True, "content": response.choices[0].message.content}
         except Exception as e:
-            print(f"GPT-4o Error: {str(e)}")
-            return ""
+            return {"success": False, "error": str(e)}
 
-    def hierarchical_reasoning(self, query: str) -> Tuple[str, dict]:
-        """Seven-stage AGI reasoning pipeline"""
+    def call_openrouter(self, model: str, prompt: str) -> dict:
+        """OpenRouter caller"""
         try:
-            # Stage 1: Conceptual Decomposition (Mistral)
-            decomposition = self._call_mistral(
-                f"""Decompose query into components: "{query}"
-                Output format:
-                - Primary Intent
-                - Implicit Assumptions
-                - Required Knowledge
-                - Potential Biases"""
-            )
-            
-            # Stage 2: Deep Analysis (GPT-4o)
-            analysis = self._call_gpt4o(
-                f"""Analyze this query using first principles:
-                {query}
-                Decomposition: {decomposition}"""
-            )
-            
-            # Stage 3: Contextual Grounding (Qwen)
-            context = self.call_model(
-                "Qwen",
-                f"""Generate multimodal context for: {query}"""
-            )
-            
-            # Stage 4: Critical Evaluation (Mistral)
-            critique = self._call_mistral(
-                f"""Critique this analysis for logical soundness:
-                Query: {query}
-                Analysis: {analysis}
-                Context: {context}"""
-            )
-            
-            # Stage 5: Ethical Consideration (Llama)
-            ethics = self.call_model(
-                "Llama",
-                f"""Analyze ethical dimensions of responding to:
-                {query}
-                Provide concrete recommendations"""
-            )
-            
-            # Stage 6: Response Synthesis (Gemini)
-            synthesis = self.call_model(
-                "Gemini",
-                f"""Synthesize final response considering:
-                {json.dumps({
-                    "query": query,
-                    "analysis": analysis,
-                    "critique": critique,
-                    "ethics": ethics
-                }, indent=2)}"""
-            )
-            
-            # Stage 7: Validation & Refinement (GPT-4o)
-            validation = self._call_gpt4o(
-                f"""Validate and refine this response:
-                Original Query: {query}
-                Proposed Response: {synthesis}
-                Provide final optimized version"""
+            response = requests.post(
+                "https://openrouter.ai/api/v1/chat/completions",
+                headers=self.headers["openrouter"],
+                json={
+                    "model": model,
+                    "messages": [{"role": "user", "content": prompt}],
+                    "temperature": 0.7,
+                    "max_tokens": 3000
+                },
+                timeout=30
             )
-            
-            # Store cognitive process
-            self.thinking_steps = [
-                ("Decomposition", decomposition),
-                ("Analysis", analysis),
-                ("Context", context),
-                ("Critique", critique),
-                ("Ethics", ethics),
-                ("Synthesis", synthesis),
-                ("Validation", validation)
-            ]
-            
-            return validation, {
-                "components": self.extract_structured_data(decomposition),
-                "analysis": self.extract_structured_data(analysis),
-                "validation": self.extract_structured_data(validation)
-            }
-        
+            if response.status_code == 200:
+                return {"success": True, "content": response.json()['choices'][0]['message']['content']}
+            return {"success": False, "error": f"API Error {response.status_code}"}
         except Exception as e:
-            print(f"Reasoning Error: {str(e)}")
-            return "Cognitive processing failed", {}
+            return {"success": False, "error": str(e)}
 
-    def extract_structured_data(self, text: str) -> dict:
-        """Advanced text parsing with fallback"""
+    def analyze_query(self, query: str) -> dict:
+        """Full analysis workflow with error resilience"""
         try:
-            # Attempt JSON extraction
-            json_match = re.search(r'\{.*\}', text, re.DOTALL)
-            if json_match:
-                return json.loads(json_match.group(0))
-            
-            # Fallback pattern matching
-            structured_data = {}
-            patterns = {
-                "intent": r"Intent:\s*(.*?)\n",
-                "components": r"Components:\s*(.*?)\n",
-                "analysis": r"Analysis:\s*(.*?)\n",
-                "recommendations": r"Recommendations:\s*(.*)"
+            # Step 1: Multi-model decomposition
+            decompositions = {}
+            for model in self.model_config["decomposition"]:
+                result = self.call_model(model, f"Decompose this query: {query}")
+                if result["success"]:
+                    decompositions[model] = self.safe_json_parse(result["content"])
+            
+            # Step 2: GPT-4 analysis
+            analysis_prompt = f"""Analyze this query based on decompositions: 
+            {json.dumps(decompositions, indent=2)}
+            
+            Query: {query}
+            Provide detailed analysis in JSON format."""
+            
+            analysis = self.call_model("gpt-4o", analysis_prompt)
+            if not analysis["success"]:
+                return {"error": "Analysis failed", "details": analysis["error"]}
+            
+            parsed_analysis = self.safe_json_parse(analysis["content"])
+            
+            # Step 3: Cross-model critique
+            critiques = {}
+            for model in self.model_config["critique"]:
+                critique_prompt = f"""Critique this analysis: 
+                {json.dumps(parsed_analysis, indent=2)}
+                
+                Provide structured feedback in JSON format."""
+                
+                result = self.call_model(model, critique_prompt)
+                if result["success"]:
+                    critiques[model] = self.safe_json_parse(result["content"])
+            
+            # Step 4: Final synthesis
+            synthesis_prompt = f"""Synthesize final response considering:
+            Analysis: {json.dumps(parsed_analysis, indent=2)}
+            Critiques: {json.dumps(critiques, indent=2)}
+            
+            Provide comprehensive response with confidence score."""
+            
+            synthesis = self.call_model("gemini-2.0-pro-exp-02-05", synthesis_prompt)
+            
+            return {
+                "decompositions": decompositions,
+                "analysis": parsed_analysis,
+                "critiques": critiques,
+                "synthesis": synthesis["content"] if synthesis["success"] else "Synthesis failed",
+                "success": True
             }
             
-            for key, pattern in patterns.items():
-                match = re.search(pattern, text)
-                if match:
-                    structured_data[key] = match.group(1).strip()
-            
-            return structured_data if structured_data else {"content": text}
-        
         except Exception as e:
-            print(f"Parsing Error: {str(e)}")
-            return {"error": "Failed to parse response"}
+            return {"success": False, "error": str(e)}
 
-    def visualize_thought_process(self) -> str:
-        """Interactive process visualization"""
-        vis = ["## Cognitive Process Breakdown\n"]
-        for step, content in self.thinking_steps:
-            vis.append(f"""
-            <div class="process-step">
-                <h3>{step}</h3>
-                <pre>{content}</pre>
-            </div>
-            """)
-        return "\n".join(vis)
-
-def create_agi_interface():
+def create_interface():
     try:
-        agi = CognitiveArchitecture()
+        analyzer = CognitiveArchitecture()
     except ValueError as e:
         return gr.Blocks().launch(error_message=str(e))
     
-    with gr.Blocks(title="AGI Thinking Framework", theme=gr.themes.Soft(), css="""
-        .process-step { margin: 20px 0; padding: 15px; border: 1px solid #e0e0e0; border-radius: 8px; }
-        .process-step h3 { color: #2b6cb0; margin: 0 0 10px 0; font-size: 1.1em; }
-        pre { 
-            white-space: pre-wrap; 
-            background: #f8f9fa; 
-            padding: 15px; 
-            border-radius: 6px; 
-            border: 1px solid #eee;
-            font-family: monospace;
-        }
+    with gr.Blocks(title="AI Analysis Suite", theme=gr.themes.Soft(), css="""
+        .analysis-section { margin: 15px; padding: 15px; border: 1px solid #eee; border-radius: 8px; }
+        pre { white-space: pre-wrap; background: #f8f9fa; padding: 10px; }
+        .error { color: #dc3545; background: #ffeef0; padding: 10px; border-radius: 5px; }
         """) as demo:
-        gr.Markdown("# 🧠 Advanced AGI Reasoning System")
+        
+        gr.Markdown("# 🧠 Advanced AI Analysis System")
         
         with gr.Row():
-            input_box = gr.Textbox(label="Input Query", 
-                                 placeholder="Enter your complex request...",
-                                 lines=3)
-            process_btn = gr.Button("Start Cognitive Processing", variant="primary")
+            input_box = gr.Textbox(label="Input Query", placeholder="Enter your query...", lines=3)
+            submit_btn = gr.Button("Analyze", variant="primary")
         
-        output = gr.Markdown()
-        process_visual = gr.HTML()
+        output_area = gr.Markdown()
+        debug_info = gr.JSON(label="Analysis Details")
         
         def process_query(query):
             start_time = time.time()
-            final, _ = agi.hierarchical_reasoning(query)
-            process_time = time.time() - start_time
+            result = analyzer.analyze_query(query)
+            duration = time.time() - start_time
             
-            return (
-                f"## Optimized Response\n{final}\n\n"
-                f"**Processing Time**: {process_time:.2f}s\n"
-                f"**Cognitive Steps Executed**: {len(agi.thinking_steps)}",
-                agi.visualize_thought_process()
-            )
+            if not result.get("success"):
+                return f"## ❌ Analysis Error\n{result.get('error', 'Unknown error')}", {}
+            
+            formatted_output = f"""
+            ## Analysis Result
+            {result.get('synthesis', 'No synthesis available')}
+            
+            **Processing Time**: {duration:.2f}s
+            **Models Used**: {len(result['decompositions'])} decomposition, {len(result['critiques'])} critique
+            """
+            
+            return formatted_output, {
+                "decompositions": result["decompositions"],
+                "analysis": result["analysis"],
+                "critiques": result["critiques"]
+            }
         
-        process_btn.click(
-            fn=process_query,
+        submit_btn.click(
+            process_query,
             inputs=input_box,
-            outputs=[output, process_visual]
+            outputs=[output_area, debug_info]
         )
-        
+    
     return demo
 
 if __name__ == "__main__":
-    create_agi_interface().launch(server_port=7860)
+    create_interface().launch(server_port=7860)