tahamajs's picture
download
raw
16.7 kB
#!/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__)
@dataclass
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.