Buckets:
tahamajs/Sysmem2_in_AI / ComputerAssignments /CA5_geometric_reasoning /src /agents /geometry_agent.py
| """ | |
| Geometry Solver Agent using LangChain | |
| عامل حل‌کننده هندسه با استفاده از LangChain | |
| This module implements a specialized agent for solving geometric problems | |
| using LangChain's agent framework with custom tools and memory. | |
| """ | |
| import logging | |
| from typing import Dict, List, Any, Optional | |
| from datetime import datetime | |
| from langchain.agents import AgentExecutor, create_conversational_agent | |
| from langchain.memory import ConversationBufferMemory | |
| from langchain.tools import Tool | |
| from langchain.prompts import PromptTemplate | |
| from langchain.schema import BaseMessage, HumanMessage, AIMessage | |
| from langchain.llms.base import BaseLLM | |
| logger = logging.getLogger(__name__) | |
| class GeometrySolverAgent: | |
| """ | |
| Specialized agent for solving geometric problems | |
| عامل تخصصی برای حل مسائل هندسی | |
| """ | |
| def __init__( | |
| self, llm: BaseLLM, memory: ConversationBufferMemory, config: Dict[str, Any] | |
| ): | |
| """ | |
| Initialize the geometry solver agent | |
| Args: | |
| llm: Language model instance | |
| memory: Conversation memory | |
| config: Agent configuration | |
| """ | |
| self.llm = llm | |
| self.memory = memory | |
| self.config = config | |
| # Agent settings | |
| self.name = config.get("name", "GeometrySolverAgent") | |
| self.description = config.get( | |
| "description", "Specialized agent for solving geometric problems" | |
| ) | |
| self.tools = config.get("tools", []) | |
| # Agent parameters | |
| self.max_iterations = config.get("settings", {}).get("max_iterations", 15) | |
| self.temperature = config.get("settings", {}).get("temperature", 0.1) | |
| self.verbose = config.get("settings", {}).get("verbose", True) | |
| self.return_intermediate_steps = config.get("settings", {}).get( | |
| "return_intermediate_steps", True | |
| ) | |
| logger.info(f"Geometry Solver Agent '{self.name}' initialized") | |
| def create_agent(self) -> AgentExecutor: | |
| """ | |
| Create the conversational agent | |
| Returns: | |
| Configured agent executor | |
| """ | |
| try: | |
| # Create system prompt | |
| system_prompt = self._create_system_prompt() | |
| # Create prompt template | |
| prompt = PromptTemplate( | |
| template=system_prompt, | |
| input_variables=["input", "agent_scratchpad", "chat_history"], | |
| ) | |
| # Create agent | |
| agent = create_conversational_agent( | |
| llm=self.llm, tools=self.tools, prompt=prompt, verbose=self.verbose | |
| ) | |
| logger.info("Conversational agent created successfully") | |
| return agent | |
| except Exception as e: | |
| logger.error(f"Error creating agent: {e}") | |
| raise | |
| def _create_system_prompt(self) -> str: | |
| """ | |
| Create the system prompt for the agent | |
| Returns: | |
| System prompt string | |
| """ | |
| prompt = """ | |
| شما یک متخصص در هندسه المپیادی هستید که با استفاده از ابزارهای مختلف مسائل هندسی را حل می‌کنید. | |
| قابلیت‌های شما: | |
| - حل مسائل هندسه تحلیلی (coordinate geometry) | |
| - حل مسائل هندسه ترکیبی (synthetic geometry) | |
| - جستجوی قضایای مرتبط | |
| - تأیید اثبات‌ها | |
| - تولید نمودارهای هندسی | |
| - جستجو در پایگاه دانش | |
| راهنمای حل مسئله: | |
| 1. ابتدا مسئله را به دقت بخوانید و تحلیل کنید | |
| 2. نوع مسئله را تشخیص دهید (مثلث، دایره، چندضلعی، و غیره) | |
| 3. از ابزارهای مناسب استفاده کنید | |
| 4. راه‌حل گام به گام ارائه دهید | |
| 5. اثبات کنید که راه‌حل درست است | |
| 6. در صورت نیاز نمودار رسم کنید | |
| نکات مهم: | |
| - همیشه پاسخ خود را به فارسی ارائه دهید | |
| - از نمادهای ریاضی مناسب استفاده کنید | |
| - مراحل حل را به وضوح توضیح دهید | |
| - در صورت عدم اطمینان، از ابزارهای موجود استفاده کنید | |
| تاریخچه گفتگو: | |
| {chat_history} | |
| ورودی کاربر: {input} | |
| فکر کردن: {agent_scratchpad} | |
| """ | |
| return prompt | |
| def solve_problem( | |
| self, problem_statement: str, tools: List[Tool] | |
| ) -> Dict[str, Any]: | |
| """ | |
| Solve a geometric problem using the agent | |
| Args: | |
| problem_statement: The problem to solve | |
| tools: List of available tools | |
| Returns: | |
| Dictionary containing the solution and metadata | |
| """ | |
| try: | |
| # Create agent with tools | |
| agent = self.create_agent() | |
| # Create agent executor | |
| agent_executor = AgentExecutor.from_agent_and_tools( | |
| agent=agent, | |
| tools=tools, | |
| memory=self.memory, | |
| verbose=self.verbose, | |
| max_iterations=self.max_iterations, | |
| handle_parsing_errors=True, | |
| return_intermediate_steps=self.return_intermediate_steps, | |
| ) | |
| # Execute the agent | |
| result = agent_executor.invoke( | |
| {"input": f"لطفاً این مسئله هندسی را حل کنید: {problem_statement}"} | |
| ) | |
| # Extract solution information | |
| solution = { | |
| "method": "langchain_agent", | |
| "agent_name": self.name, | |
| "solution": result.get("output", ""), | |
| "intermediate_steps": result.get("intermediate_steps", []), | |
| "tools_used": self._extract_tools_used( | |
| result.get("intermediate_steps", []) | |
| ), | |
| "confidence": self._calculate_confidence(result), | |
| "timestamp": datetime.now().isoformat(), | |
| "memory_used": ( | |
| len(self.memory.chat_memory.messages) | |
| if hasattr(self.memory, "chat_memory") | |
| else 0 | |
| ), | |
| } | |
| logger.info(f"Problem solved using agent '{self.name}'") | |
| return solution | |
| except Exception as e: | |
| logger.error(f"Error solving problem with agent: {e}") | |
| return { | |
| "method": "langchain_agent", | |
| "agent_name": self.name, | |
| "error": str(e), | |
| "timestamp": datetime.now().isoformat(), | |
| } | |
| def _extract_tools_used(self, intermediate_steps: List[Any]) -> List[str]: | |
| """ | |
| Extract tools used from intermediate steps | |
| Args: | |
| intermediate_steps: List of intermediate steps | |
| Returns: | |
| List of tool names used | |
| """ | |
| tools_used = [] | |
| for step in intermediate_steps: | |
| if isinstance(step, dict) and "action" in step: | |
| tool_name = step["action"].get("tool", "") | |
| if tool_name and tool_name not in tools_used: | |
| tools_used.append(tool_name) | |
| return tools_used | |
| def _calculate_confidence(self, result: Dict[str, Any]) -> float: | |
| """ | |
| Calculate confidence score for the solution | |
| Args: | |
| result: Agent execution result | |
| Returns: | |
| Confidence score between 0 and 1 | |
| """ | |
| try: | |
| confidence = 0.5 # Base confidence | |
| # Check if solution is complete | |
| solution = result.get("output", "") | |
| if solution and len(solution) > 50: | |
| confidence += 0.2 | |
| # Check if tools were used effectively | |
| intermediate_steps = result.get("intermediate_steps", []) | |
| if intermediate_steps: | |
| confidence += 0.2 | |
| # Check for mathematical content | |
| if any( | |
| keyword in solution.lower() | |
| for keyword in ["قضیه", "اثبات", "محاسبه", "نتیجه"] | |
| ): | |
| confidence += 0.1 | |
| return min(confidence, 1.0) | |
| except Exception as e: | |
| logger.error(f"Error calculating confidence: {e}") | |
| return 0.5 | |
| def get_agent_info(self) -> Dict[str, Any]: | |
| """ | |
| Get information about the agent | |
| Returns: | |
| Dictionary containing agent information | |
| """ | |
| return { | |
| "name": self.name, | |
| "description": self.description, | |
| "tools": self.tools, | |
| "max_iterations": self.max_iterations, | |
| "temperature": self.temperature, | |
| "verbose": self.verbose, | |
| "memory_type": type(self.memory).__name__, | |
| "memory_size": ( | |
| len(self.memory.chat_memory.messages) | |
| if hasattr(self.memory, "chat_memory") | |
| else 0 | |
| ), | |
| } | |
| def clear_memory(self): | |
| """Clear the agent's memory""" | |
| try: | |
| if hasattr(self.memory, "clear"): | |
| self.memory.clear() | |
| elif hasattr(self.memory, "chat_memory"): | |
| self.memory.chat_memory.clear() | |
| logger.info("Agent memory cleared") | |
| except Exception as e: | |
| logger.error(f"Error clearing memory: {e}") | |
| def add_to_memory(self, message: str, is_human: bool = True): | |
| """ | |
| Add a message to the agent's memory | |
| Args: | |
| message: Message to add | |
| is_human: Whether the message is from human or AI | |
| """ | |
| try: | |
| if hasattr(self.memory, "chat_memory"): | |
| if is_human: | |
| self.memory.chat_memory.add_user_message(message) | |
| else: | |
| self.memory.chat_memory.add_ai_message(message) | |
| logger.info(f"Message added to memory: {message[:50]}...") | |
| except Exception as e: | |
| logger.error(f"Error adding message to memory: {e}") | |
| def get_memory_summary(self) -> Dict[str, Any]: | |
| """ | |
| Get a summary of the agent's memory | |
| Returns: | |
| Dictionary containing memory summary | |
| """ | |
| try: | |
| summary = { | |
| "total_messages": 0, | |
| "human_messages": 0, | |
| "ai_messages": 0, | |
| "memory_type": type(self.memory).__name__, | |
| } | |
| if hasattr(self.memory, "chat_memory"): | |
| messages = self.memory.chat_memory.messages | |
| summary["total_messages"] = len(messages) | |
| for message in messages: | |
| if isinstance(message, HumanMessage): | |
| summary["human_messages"] += 1 | |
| elif isinstance(message, AIMessage): | |
| summary["ai_messages"] += 1 | |
| return summary | |
| except Exception as e: | |
| logger.error(f"Error getting memory summary: {e}") | |
| return {} | |
| def update_config(self, new_config: Dict[str, Any]): | |
| """ | |
| Update agent configuration | |
| Args: | |
| new_config: New configuration parameters | |
| """ | |
| try: | |
| self.config.update(new_config) | |
| # Update agent parameters | |
| if "settings" in new_config: | |
| settings = new_config["settings"] | |
| self.max_iterations = settings.get( | |
| "max_iterations", self.max_iterations | |
| ) | |
| self.temperature = settings.get("temperature", self.temperature) | |
| self.verbose = settings.get("verbose", self.verbose) | |
| self.return_intermediate_steps = settings.get( | |
| "return_intermediate_steps", self.return_intermediate_steps | |
| ) | |
| logger.info("Agent configuration updated") | |
| except Exception as e: | |
| logger.error(f"Error updating agent configuration: {e}") | |
| def export_conversation(self) -> List[Dict[str, Any]]: | |
| """ | |
| Export the conversation history | |
| Returns: | |
| List of conversation messages | |
| """ | |
| try: | |
| conversation = [] | |
| if hasattr(self.memory, "chat_memory"): | |
| for message in self.memory.chat_memory.messages: | |
| msg_dict = { | |
| "type": "human" if isinstance(message, HumanMessage) else "ai", | |
| "content": message.content, | |
| "timestamp": datetime.now().isoformat(), | |
| } | |
| conversation.append(msg_dict) | |
| return conversation | |
| except Exception as e: | |
| logger.error(f"Error exporting conversation: {e}") | |
| return [] | |
Xet Storage Details
- Size:
- 12.8 kB
- Xet hash:
- d566731201cc15a3f163d886f6eff34819ade3889257789df36509e3ae92bae3
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.