Buckets:
tahamajs/Sysmem2_in_AI / ComputerAssignments /CA5_geometric_reasoning /src /agents /geometric_agent.py
| #!/usr/bin/env python3 | |
| import os | |
| import json | |
| import logging | |
| from typing import List, Dict, Any, Optional, Tuple | |
| from dataclasses import dataclass | |
| from abc import ABC, abstractmethod | |
| from langchain_google_genai import ChatGoogleGenerativeAI | |
| from langchain_core.messages import HumanMessage, AIMessage, SystemMessage | |
| from langchain_core.tools import Tool | |
| from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder | |
| from langchain_core.output_parsers import StrOutputParser | |
| from langchain_core.runnables import RunnablePassthrough, RunnableLambda | |
| from langgraph.graph import StateGraph, END | |
| from langgraph.prebuilt import ToolNode | |
| from langgraph.checkpoint.memory import MemorySaver | |
| import numpy as np | |
| import sympy as sp | |
| from sympy.geometry import Point, Line, Circle, Triangle | |
| import matplotlib.pyplot as plt | |
| import networkx as nx | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| class AgentState: | |
| problem_statement: str | |
| current_step: int | |
| reasoning_trace: List[str] | |
| geometric_objects: Dict[str, Any] | |
| theorems_applied: List[str] | |
| proof_steps: List[str] | |
| confidence: float | |
| error_message: Optional[str] = None | |
| is_complete: bool = False | |
| class GeometricToolkit: | |
| def __init__(self): | |
| self.theorem_database = self._initialize_theorems() | |
| self.construction_methods = self._initialize_constructions() | |
| def _initialize_theorems(self) -> Dict[str, Dict]: | |
| return { | |
| "pythagorean": { | |
| "statement": "In a right triangle, a² + b² = c²", | |
| "conditions": ["right_triangle"], | |
| "conclusions": ["pythagorean_relation"], | |
| }, | |
| "angle_sum": { | |
| "statement": "Sum of angles in triangle is 180°", | |
| "conditions": ["triangle"], | |
| "conclusions": ["angle_sum_180"], | |
| }, | |
| "circumcenter": { | |
| "statement": "Perpendicular bisectors meet at circumcenter", | |
| "conditions": ["triangle"], | |
| "conclusions": ["circumcenter_exists"], | |
| }, | |
| "inscribed_angle": { | |
| "statement": "Inscribed angle is half central angle", | |
| "conditions": ["circle", "inscribed_angle"], | |
| "conclusions": ["angle_ratio_1_2"], | |
| }, | |
| "power_of_point": { | |
| "statement": "Power of point P wrt circle = |PO|² - r²", | |
| "conditions": ["point", "circle"], | |
| "conclusions": ["power_formula"], | |
| }, | |
| } | |
| def _initialize_constructions(self) -> Dict[str, str]: | |
| return { | |
| "perpendicular_bisector": "Construct perpendicular bisector of segment", | |
| "angle_bisector": "Construct angle bisector", | |
| "parallel_line": "Construct line parallel to given line through point", | |
| "circumcircle": "Construct circumcircle of triangle", | |
| "incircle": "Construct incircle of triangle", | |
| } | |
| def analyze_problem_structure(self, problem: str) -> Dict[str, Any]: | |
| analysis = { | |
| "geometric_objects": [], | |
| "relationships": [], | |
| "constraints": [], | |
| "goal": "", | |
| "difficulty": "unknown", | |
| } | |
| keywords = { | |
| "triangle": ["triangle", "triangular"], | |
| "circle": ["circle", "circular", "circumference"], | |
| "angle": ["angle", "degree", "radian"], | |
| "line": ["line", "segment", "ray"], | |
| "point": ["point", "vertex", "intersection"], | |
| "parallel": ["parallel", "parallelism"], | |
| "perpendicular": ["perpendicular", "right angle", "orthogonal"], | |
| "similar": ["similar", "similarity"], | |
| "congruent": ["congruent", "congruence"], | |
| } | |
| problem_lower = problem.lower() | |
| for obj_type, terms in keywords.items(): | |
| if any(term in problem_lower for term in terms): | |
| analysis["geometric_objects"].append(obj_type) | |
| olympiad_indicators = ["prove", "construct", "find", "show", "demonstrate"] | |
| if any(indicator in problem_lower for indicator in olympiad_indicators): | |
| analysis["difficulty"] = "olympiad" | |
| elif "calculate" in problem_lower or "find" in problem_lower: | |
| analysis["difficulty"] = "high_school" | |
| else: | |
| analysis["difficulty"] = "elementary" | |
| return analysis | |
| def suggest_theorems(self, problem_analysis: Dict[str, Any]) -> List[str]: | |
| suggested = [] | |
| objects = problem_analysis["geometric_objects"] | |
| for theorem_name, theorem_info in self.theorem_database.items(): | |
| conditions = theorem_info["conditions"] | |
| if any(obj in objects for obj in conditions): | |
| suggested.append(theorem_name) | |
| return suggested | |
| def apply_theorem( | |
| self, theorem_name: str, context: Dict[str, Any] | |
| ) -> Dict[str, Any]: | |
| if theorem_name not in self.theorem_database: | |
| return {"success": False, "error": f"Theorem {theorem_name} not found"} | |
| theorem = self.theorem_database[theorem_name] | |
| conditions_met = all( | |
| condition in context.get("geometric_objects", []) | |
| for condition in theorem["conditions"] | |
| ) | |
| if not conditions_met: | |
| return { | |
| "success": False, | |
| "error": f"Conditions not met for theorem {theorem_name}", | |
| } | |
| return { | |
| "success": True, | |
| "theorem": theorem_name, | |
| "statement": theorem["statement"], | |
| "conclusions": theorem["conclusions"], | |
| "confidence": 0.8, | |
| } | |
| def construct_geometric_object( | |
| self, construction_type: str, parameters: Dict[str, Any] | |
| ) -> Dict[str, Any]: | |
| if construction_type not in self.construction_methods: | |
| return { | |
| "success": False, | |
| "error": f"Unknown construction: {construction_type}", | |
| } | |
| return { | |
| "success": True, | |
| "construction": construction_type, | |
| "description": self.construction_methods[construction_type], | |
| "parameters": parameters, | |
| "result": f"Constructed {construction_type} successfully", | |
| } | |
| def verify_proof_step(self, step: str, context: Dict[str, Any]) -> Dict[str, Any]: | |
| verification_keywords = ["therefore", "hence", "thus", "so", "consequently"] | |
| has_conclusion = any( | |
| keyword in step.lower() for keyword in verification_keywords | |
| ) | |
| return { | |
| "valid": has_conclusion and len(step.split()) > 3, | |
| "confidence": 0.7 if has_conclusion else 0.3, | |
| "feedback": ( | |
| "Step appears logically sound" | |
| if has_conclusion | |
| else "Step needs clearer conclusion" | |
| ), | |
| } | |
| class GeminiGeometricReasoner: | |
| def __init__(self, api_key: Optional[str] = None): | |
| self.api_key = api_key or os.getenv("GOOGLE_API_KEY") | |
| if not self.api_key: | |
| raise ValueError("Google API key not provided") | |
| self.llm = ChatGoogleGenerativeAI( | |
| model="gemini-1.5-pro", | |
| temperature=0.1, | |
| max_retries=3, | |
| google_api_key=self.api_key, | |
| ) | |
| self.toolkit = GeometricToolkit() | |
| self._setup_tools() | |
| def _setup_tools(self): | |
| self.tools = [ | |
| Tool( | |
| name="analyze_problem", | |
| description="Analyze the structure of a geometric problem", | |
| func=self.toolkit.analyze_problem_structure, | |
| ), | |
| Tool( | |
| name="suggest_theorems", | |
| description="Suggest relevant theorems for a problem", | |
| func=self.toolkit.suggest_theorems, | |
| ), | |
| Tool( | |
| name="apply_theorem", | |
| description="Apply a specific theorem to current context", | |
| func=self.toolkit.apply_theorem, | |
| ), | |
| Tool( | |
| name="construct_object", | |
| description="Perform geometric construction", | |
| func=self.toolkit.construct_geometric_object, | |
| ), | |
| Tool( | |
| name="verify_step", | |
| description="Verify correctness of a proof step", | |
| func=self.toolkit.verify_proof_step, | |
| ), | |
| ] | |
| self.llm_with_tools = self.llm.bind_tools(self.tools) | |
| def create_reasoning_prompt(self) -> ChatPromptTemplate: | |
| return ChatPromptTemplate.from_messages( | |
| [ | |
| SystemMessage( | |
| content= | |
| ), | |
| MessagesPlaceholder(variable_name="chat_history"), | |
| HumanMessage(content="{problem_statement}"), | |
| ] | |
| ) | |
| def reason_about_problem( | |
| self, problem: str, context: Dict[str, Any] = None | |
| ) -> Dict[str, Any]: | |
| try: | |
| prompt = self.create_reasoning_prompt() | |
| analysis = self.toolkit.analyze_problem_structure(problem) | |
| suggested_theorems = self.toolkit.suggest_theorems(analysis) | |
| reasoning_context = { | |
| "problem": problem, | |
| "analysis": analysis, | |
| "suggested_theorems": suggested_theorems, | |
| "context": context or {}, | |
| } | |
| chain = prompt | self.llm_with_tools | StrOutputParser() | |
| response = chain.invoke( | |
| { | |
| "problem_statement": f | |
| "chat_history": [], | |
| } | |
| ) | |
| return { | |
| "success": True, | |
| "reasoning": response, | |
| "analysis": analysis, | |
| "suggested_theorems": suggested_theorems, | |
| "confidence": 0.8, | |
| } | |
| except Exception as e: | |
| logger.error(f"Error in reasoning: {str(e)}") | |
| return {"success": False, "error": str(e), "confidence": 0.0} | |
| def generate_proof_steps(self, problem: str, approach: str) -> List[str]: | |
| try: | |
| prompt = f | |
| response = self.llm.invoke([HumanMessage(content=prompt)]) | |
| steps = [] | |
| lines = response.content.split("\n") | |
| for line in lines: | |
| line = line.strip() | |
| if line and (line[0].isdigit() or line.startswith("Step")): | |
| steps.append(line) | |
| return steps if steps else ["Proof steps could not be generated"] | |
| except Exception as e: | |
| logger.error(f"Error generating proof steps: {str(e)}") | |
| return [f"Error: {str(e)}"] | |
| class LangGraphGeometricWorkflow: | |
| def __init__(self, gemini_reasoner: GeminiGeometricReasoner): | |
| self.reasoner = gemini_reasoner | |
| self.memory = MemorySaver() | |
| self.graph = self._build_graph() | |
| def _build_graph(self) -> StateGraph: | |
| graph = StateGraph(AgentState) | |
| graph.add_node("analyze", self._analyze_problem) | |
| graph.add_node("plan", self._plan_solution) | |
| graph.add_node("execute", self._execute_proof) | |
| graph.add_node("verify", self._verify_solution) | |
| graph.add_node("explain", self._explain_solution) | |
| graph.add_edge("analyze", "plan") | |
| graph.add_edge("plan", "execute") | |
| graph.add_edge("execute", "verify") | |
| graph.add_edge("verify", "explain") | |
| graph.add_edge("explain", END) | |
| graph.set_entry_point("analyze") | |
| return graph.compile(checkpointer=self.memory) | |
| def _analyze_problem(self, state: AgentState) -> AgentState: | |
| try: | |
| analysis = self.reasoner.toolkit.analyze_problem_structure( | |
| state.problem_statement | |
| ) | |
| state.geometric_objects = analysis | |
| state.reasoning_trace.append( | |
| f"Analyzed problem: {analysis['difficulty']} level" | |
| ) | |
| return state | |
| except Exception as e: | |
| state.error_message = f"Analysis error: {str(e)}" | |
| return state | |
| def _plan_solution(self, state: AgentState) -> AgentState: | |
| try: | |
| reasoning_result = self.reasoner.reason_about_problem( | |
| state.problem_statement | |
| ) | |
| if reasoning_result["success"]: | |
| state.reasoning_trace.append("Generated solution plan") | |
| state.confidence = reasoning_result["confidence"] | |
| else: | |
| state.error_message = reasoning_result.get("error", "Planning failed") | |
| return state | |
| except Exception as e: | |
| state.error_message = f"Planning error: {str(e)}" | |
| return state | |
| def _execute_proof(self, state: AgentState) -> AgentState: | |
| try: | |
| if state.error_message: | |
| return state | |
| proof_steps = self.reasoner.generate_proof_steps( | |
| state.problem_statement, "Systematic geometric reasoning" | |
| ) | |
| state.proof_steps = proof_steps | |
| state.reasoning_trace.append(f"Generated {len(proof_steps)} proof steps") | |
| return state | |
| except Exception as e: | |
| state.error_message = f"Execution error: {str(e)}" | |
| return state | |
| def _verify_solution(self, state: AgentState) -> AgentState: | |
| try: | |
| if state.error_message or not state.proof_steps: | |
| return state | |
| verification_results = [] | |
| for step in state.proof_steps: | |
| verification = self.reasoner.toolkit.verify_proof_step( | |
| step, state.geometric_objects | |
| ) | |
| verification_results.append(verification) | |
| avg_confidence = sum(v["confidence"] for v in verification_results) / len( | |
| verification_results | |
| ) | |
| state.confidence = avg_confidence | |
| state.reasoning_trace.append( | |
| f"Verified solution with confidence {avg_confidence:.2f}" | |
| ) | |
| return state | |
| except Exception as e: | |
| state.error_message = f"Verification error: {str(e)}" | |
| return state | |
| def _explain_solution(self, state: AgentState) -> AgentState: | |
| try: | |
| if state.error_message: | |
| state.reasoning_trace.append(f"Solution failed: {state.error_message}") | |
| return state | |
| explanation = f | |
| state.reasoning_trace.append("Generated final explanation") | |
| state.is_complete = True | |
| return state | |
| except Exception as e: | |
| state.error_message = f"Explanation error: {str(e)}" | |
| return state | |
| def solve_problem(self, problem_statement: str) -> AgentState: | |
| initial_state = AgentState( | |
| problem_statement=problem_statement, | |
| current_step=0, | |
| reasoning_trace=[], | |
| geometric_objects={}, | |
| theorems_applied=[], | |
| proof_steps=[], | |
| confidence=0.0, | |
| ) | |
| final_state = self.graph.invoke(initial_state) | |
| return final_state | |
| def main(): | |
| print("=== Advanced Geometric Reasoning Agent ===") | |
| try: | |
| reasoner = GeminiGeometricReasoner() | |
| workflow = LangGraphGeometricWorkflow(reasoner) | |
| test_problems = [ | |
| "In triangle ABC with right angle at C, prove that AB² = AC² + BC²", | |
| "Prove that the perpendicular bisectors of a triangle meet at a single point", | |
| "In circle O, prove that an inscribed angle is half the central angle subtending the same arc", | |
| ] | |
| for i, problem in enumerate(test_problems, 1): | |
| print(f"\n{'='*60}") | |
| print(f"PROBLEM {i}: {problem}") | |
| print(f"{'='*60}") | |
| result = workflow.solve_problem(problem) | |
| if result.is_complete and not result.error_message: | |
| print("✅ Problem solved successfully!") | |
| print(f"Confidence: {result.confidence:.1%}") | |
| print(f"Proof steps: {len(result.proof_steps)}") | |
| print("\nReasoning Trace:") | |
| for trace in result.reasoning_trace: | |
| print(f" • {trace}") | |
| else: | |
| print("❌ Problem solving failed") | |
| if result.error_message: | |
| print(f"Error: {result.error_message}") | |
| except Exception as e: | |
| print(f"❌ System initialization failed: {str(e)}") | |
| print("Make sure to set GOOGLE_API_KEY environment variable") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 16.7 kB
- Xet hash:
- f937d44e5f28798b0f8a28f9fbe625840729eeef89fc7da1e8e3350e1bccb93e
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.