| """ | |
| Reasoning tools for the GAIA agent. | |
| This module provides tools for complex reasoning tasks, including: | |
| - Step-by-step reasoning | |
| - Mathematical calculations | |
| - Fact verification | |
| All tools handle errors gracefully and provide detailed error messages. | |
| """ | |
| import logging | |
| import traceback | |
| import re | |
| import math | |
| import json | |
| from typing import Dict, Any, List, Optional, Union, Tuple, Callable | |
| from datetime import datetime | |
| import operator | |
| from sympy import symbols, sympify, solve, simplify | |
| import numpy as np | |
| from src.gaia.agent.config import get_model_config, get_tool_config | |
| from langchain_openai import ChatOpenAI | |
| from langchain.prompts import PromptTemplate | |
| from langchain_core.output_parsers import StrOutputParser | |
| logger = logging.getLogger("gaia_agent.tools.reasoning") | |
| class StepByStepReasoner: | |
| """Tool for performing step-by-step reasoning.""" | |
| def __init__(self, config: Optional[Dict[str, Any]] = None): | |
| """ | |
| Initialize the step-by-step reasoner. | |
| Args: | |
| config: Optional configuration dictionary | |
| """ | |
| self.model_config = config or get_model_config() | |
| self.model = ChatOpenAI( | |
| model=self.model_config.get("reasoning_model", "gpt-4o"), | |
| temperature=self.model_config.get("temperature", 0.1), | |
| max_tokens=self.model_config.get("max_tokens", 4096) | |
| ) | |
| def reason(self, question: str, context: Optional[str] = None) -> Dict[str, Any]: | |
| """ | |
| Perform step-by-step reasoning on a question. | |
| Args: | |
| question: The question to reason about | |
| context: Optional context information | |
| Returns: | |
| Dictionary containing the reasoning steps and conclusion | |
| Raises: | |
| Exception: If an error occurs during reasoning | |
| """ | |
| try: | |
| prompt_template = """You are an expert at step-by-step reasoning. | |
| Your task is to solve the following problem by breaking it down into clear, logical steps. | |
| {context_text} | |
| Question: {question} | |
| Think through this step-by-step and provide your reasoning in the following JSON format: | |
| {{ | |
| "steps": [ | |
| {{ | |
| "step_number": 1, | |
| "reasoning": "First step of reasoning", | |
| "intermediate_conclusion": "Conclusion from this step" | |
| }}, | |
| ... | |
| ], | |
| "final_conclusion": "The final answer based on all steps", | |
| "confidence": 0.95 // A number between 0 and 1 indicating your confidence | |
| }} | |
| JSON Response:""" | |
| context_text = f"Context: {context}" if context else "No additional context provided." | |
| prompt = PromptTemplate.from_template(prompt_template) | |
| chain = prompt | self.model | StrOutputParser() | |
| result = chain.invoke({ | |
| "question": question, | |
| "context_text": context_text | |
| }) | |
| parsed_result = json.loads(result) | |
| return parsed_result | |
| except Exception as e: | |
| logger.error(f"Error during step-by-step reasoning: {str(e)}") | |
| logger.error(traceback.format_exc()) | |
| raise Exception(f"Step-by-step reasoning failed: {str(e)}") | |
| def chain_of_thought(self, question: str, context: Optional[str] = None) -> str: | |
| """ | |
| Perform chain-of-thought reasoning and return a textual response. | |
| Args: | |
| question: The question to reason about | |
| context: Optional context information | |
| Returns: | |
| String containing the reasoning and conclusion | |
| Raises: | |
| Exception: If an error occurs during reasoning | |
| """ | |
| try: | |
| prompt_template = """You are an expert at chain-of-thought reasoning. | |
| Your task is to solve the following problem by thinking step-by-step. | |
| {context_text} | |
| Question: {question} | |
| Let's think through this step-by-step:""" | |
| context_text = f"Context: {context}" if context else "No additional context provided." | |
| prompt = PromptTemplate.from_template(prompt_template) | |
| chain = prompt | self.model | StrOutputParser() | |
| result = chain.invoke({ | |
| "question": question, | |
| "context_text": context_text | |
| }) | |
| return result | |
| except Exception as e: | |
| logger.error(f"Error during chain-of-thought reasoning: {str(e)}") | |
| logger.error(traceback.format_exc()) | |
| raise Exception(f"Chain-of-thought reasoning failed: {str(e)}") | |
| class MathCalculator: | |
| """Tool for performing mathematical calculations.""" | |
| def __init__(self): | |
| """Initialize the math calculator.""" | |
| self.operations = { | |
| '+': operator.add, | |
| '-': operator.sub, | |
| '*': operator.mul, | |
| '/': operator.truediv, | |
| '^': operator.pow, | |
| '**': operator.pow, | |
| '%': operator.mod, | |
| '//': operator.floordiv | |
| } | |
| def calculate(self, expression: str) -> Dict[str, Any]: | |
| """ | |
| Evaluate a mathematical expression. | |
| Args: | |
| expression: The mathematical expression to evaluate | |
| Returns: | |
| Dictionary containing the result and steps | |
| Raises: | |
| Exception: If an error occurs during calculation | |
| """ | |
| try: | |
| cleaned_expr = self._clean_expression(expression) | |
| result = float(sympify(cleaned_expr).evalf()) | |
| if result.is_integer(): | |
| result = int(result) | |
| return { | |
| "expression": expression, | |
| "result": result, | |
| "steps": [f"Evaluated expression: {cleaned_expr}", f"Result: {result}"] | |
| } | |
| except Exception as e: | |
| logger.error(f"Error calculating expression: {str(e)}") | |
| logger.error(traceback.format_exc()) | |
| raise Exception(f"Mathematical calculation failed: {str(e)}") | |
| def solve_equation(self, equation: str, variable: str = 'x') -> Dict[str, Any]: | |
| """ | |
| Solve a mathematical equation for a variable. | |
| Args: | |
| equation: The equation to solve (e.g., "x + 5 = 10") | |
| variable: The variable to solve for (default: 'x') | |
| Returns: | |
| Dictionary containing the solution and steps | |
| Raises: | |
| Exception: If an error occurs during solving | |
| """ | |
| try: | |
| if '=' in equation: | |
| left_side, right_side = equation.split('=', 1) | |
| equation_sympy = f"({left_side.strip()}) - ({right_side.strip()})" | |
| else: | |
| equation_sympy = equation | |
| var = symbols(variable) | |
| expr = sympify(equation_sympy) | |
| solutions = solve(expr, var) | |
| formatted_solutions = [] | |
| for sol in solutions: | |
| if sol.is_real: | |
| try: | |
| float_sol = float(sol) | |
| if float_sol.is_integer(): | |
| formatted_solutions.append(int(float_sol)) | |
| else: | |
| formatted_solutions.append(float_sol) | |
| except: | |
| formatted_solutions.append(str(sol)) | |
| else: | |
| formatted_solutions.append(str(sol)) | |
| return { | |
| "equation": equation, | |
| "variable": variable, | |
| "solutions": formatted_solutions, | |
| "steps": [ | |
| f"Parsed equation: {equation}", | |
| f"Rearranged to: {equation_sympy} = 0", | |
| f"Solved for {variable}", | |
| f"Solutions: {', '.join(map(str, formatted_solutions))}" | |
| ] | |
| } | |
| except Exception as e: | |
| logger.error(f"Error solving equation: {str(e)}") | |
| logger.error(traceback.format_exc()) | |
| raise Exception(f"Equation solving failed: {str(e)}") | |
| def _clean_expression(self, expression: str) -> str: | |
| """ | |
| Clean a mathematical expression for evaluation. | |
| Args: | |
| expression: The expression to clean | |
| Returns: | |
| Cleaned expression | |
| """ | |
| expression = expression.strip() | |
| replacements = { | |
| 'sin': 'math.sin', | |
| 'cos': 'math.cos', | |
| 'tan': 'math.tan', | |
| 'log': 'math.log', | |
| 'exp': 'math.exp', | |
| 'sqrt': 'math.sqrt', | |
| 'pi': 'math.pi', | |
| 'e': 'math.e' | |
| } | |
| for old, new in replacements.items(): | |
| expression = re.sub(r'\b' + old + r'\b', new, expression) | |
| return expression | |
| class FactVerifier: | |
| """Tool for verifying facts and claims.""" | |
| def __init__(self, config: Optional[Dict[str, Any]] = None): | |
| """ | |
| Initialize the fact verifier. | |
| Args: | |
| config: Optional configuration dictionary | |
| """ | |
| self.model_config = config or get_model_config() | |
| self.model = ChatOpenAI( | |
| model=self.model_config.get("reasoning_model", "gpt-4o"), | |
| temperature=self.model_config.get("temperature", 0.1), | |
| max_tokens=self.model_config.get("max_tokens", 4096) | |
| ) | |
| def verify(self, claim: str, evidence: Optional[str] = None) -> Dict[str, Any]: | |
| """ | |
| Verify a claim against evidence. | |
| Args: | |
| claim: The claim to verify | |
| evidence: Optional evidence to use for verification | |
| Returns: | |
| Dictionary containing verification results | |
| Raises: | |
| Exception: If an error occurs during verification | |
| """ | |
| try: | |
| prompt_template = """You are an expert fact checker. | |
| Your task is to verify the following claim based on the provided evidence. | |
| Claim: {claim} | |
| Evidence: {evidence} | |
| Analyze the claim and evidence carefully. Provide your verification in the following JSON format: | |
| {{ | |
| "is_verifiable": true/false, // Whether the claim can be verified with the given evidence | |
| "verification_result": "supported"/"contradicted"/"insufficient_evidence", | |
| "confidence": 0.95, // A number between 0 and 1 indicating your confidence | |
| "reasoning": "Your step-by-step reasoning process", | |
| "missing_information": "What additional information would help verify this claim" | |
| }} | |
| JSON Response:""" | |
| evidence_text = evidence if evidence else "No evidence provided." | |
| prompt = PromptTemplate.from_template(prompt_template) | |
| chain = prompt | self.model | StrOutputParser() | |
| result = chain.invoke({ | |
| "claim": claim, | |
| "evidence": evidence_text | |
| }) | |
| parsed_result = json.loads(result) | |
| return parsed_result | |
| except Exception as e: | |
| logger.error(f"Error during fact verification: {str(e)}") | |
| logger.error(traceback.format_exc()) | |
| raise Exception(f"Fact verification failed: {str(e)}") | |
| def compare_claims(self, claim1: str, claim2: str) -> Dict[str, Any]: | |
| """ | |
| Compare two claims for consistency. | |
| Args: | |
| claim1: The first claim | |
| claim2: The second claim | |
| Returns: | |
| Dictionary containing comparison results | |
| Raises: | |
| Exception: If an error occurs during comparison | |
| """ | |
| try: | |
| prompt_template = """You are an expert at analyzing logical consistency. | |
| Your task is to compare the following two claims and determine if they are consistent, contradictory, or unrelated. | |
| Claim 1: {claim1} | |
| Claim 2: {claim2} | |
| Analyze the claims carefully. Provide your analysis in the following JSON format: | |
| {{ | |
| "relationship": "consistent"/"contradictory"/"unrelated", | |
| "confidence": 0.95, // A number between 0 and 1 indicating your confidence | |
| "reasoning": "Your step-by-step reasoning process", | |
| "implications": "What the relationship between these claims implies" | |
| }} | |
| JSON Response:""" | |
| prompt = PromptTemplate.from_template(prompt_template) | |
| chain = prompt | self.model | StrOutputParser() | |
| result = chain.invoke({ | |
| "claim1": claim1, | |
| "claim2": claim2 | |
| }) | |
| parsed_result = json.loads(result) | |
| return parsed_result | |
| except Exception as e: | |
| logger.error(f"Error during claim comparison: {str(e)}") | |
| logger.error(traceback.format_exc()) | |
| raise Exception(f"Claim comparison failed: {str(e)}") | |
| def create_step_by_step_reasoner() -> StepByStepReasoner: | |
| """Create a step-by-step reasoner instance.""" | |
| return StepByStepReasoner() | |
| def create_math_calculator() -> MathCalculator: | |
| """Create a math calculator instance.""" | |
| return MathCalculator() | |
| def create_fact_verifier() -> FactVerifier: | |
| """Create a fact verifier instance.""" | |
| return FactVerifier() |