BuddyMath / smart_solver.py
dotandru's picture
Fix: Clean production deployment with sse-starlette
9d29c62
Raw
History Blame
31.5 kB
# smart_solver.py - V7.4 (TYPED SIGNED STEPS + DOMAIN-AWARE CONTRACTS)
import re
import hashlib
import logging
import sympy
from sympy import symbols, Eq, solve, sympify, diff, latex, Symbol, simplify, trigsimp
from sympy import srepr, default_sort_key
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Tuple, Union, Optional
from dataclasses import dataclass
import copy
from domain.step_types import StepType, SignedStep
logger = logging.getLogger(__name__)
# ==================== V7.3: TYPED ACTION CONTRACT ====================
@dataclass
class ActionContext:
"""
V7.3: Typed contract between Orchestrator and Solver.
Replaces generic dict โ€” unknown fields cause AttributeError, not silent bugs.
Add new fields here as the system learns new problem types.
"""
center: Optional[Tuple[float, float]] = None
radius: Optional[float] = None
point_a: Optional[Tuple[float, float]] = None
# Future: triangle_vertices, line_coefficients, etc.
# ==================== V7.2: DETERMINISTIC SIGNATURE ENGINE ====================
def sign_step(expression: Union[str, list], problem_id: str, step_id: str) -> SignedStep:
"""
V7.4: Creates a SHA256-signed algebraic step as a typed SignedStep.
Uses sympy.srepr for canonical representation to guarantee deterministic hashing.
Sorts list results to ensure hash stability regardless of SymPy output order.
step_type = ALGEBRAIC โ€” ConsistencyGate will validate via sympy.sympify.
payload = raw expression list or string (semantic truth for validator).
expression = display string injected into renderer placeholders.
"""
if isinstance(expression, list):
# Sort for canonical order (critical for hash determinism)
parsed_exprs = sorted(
[sympy.sympify(e) for e in expression],
key=default_sort_key
)
canonical = str([srepr(e) for e in parsed_exprs])
expr_str = " OR ".join(str(e) for e in parsed_exprs)
payload = [str(e) for e in parsed_exprs]
else:
parsed = sympy.sympify(expression)
canonical = srepr(parsed)
expr_str = str(parsed)
payload = expr_str
raw = f"{canonical}{problem_id}{step_id}"
sig = hashlib.sha256(raw.encode()).hexdigest()
logger.info(f"[SIGNATURE] Signed step '{step_id}': hash={sig[:12]}...")
return SignedStep(
id=step_id,
expression=expr_str,
payload=payload,
step_type=StepType.ALGEBRAIC,
hash=sig,
)
def sign_step_geometry(labels: list, problem_id: str, step_id: str) -> SignedStep:
"""
V7.4: Typed signer for geometry results (strings, not SymPy expressions).
step_type = GEOMETRY โ€” ConsistencyGate validates structurally, NOT via sympify.
payload = original labels list (semantic truth: list[str] of points/distances).
expression = display string for renderer injection (pipe-separated, human-readable).
CTO note: expression is display-only. payload is the authoritative structure
the validator uses to verify integrity. Never pass expression to sympy.
"""
canonical = str(sorted(str(l) for l in labels))
sig = hashlib.sha256(f"{canonical}{problem_id}{step_id}".encode()).hexdigest()
expr_str = " | ".join(str(l) for l in labels)
logger.info(f"[SIGNATURE] Signed geometry step '{step_id}': hash={sig[:12]}...")
return SignedStep(
id=step_id,
expression=expr_str,
payload=list(labels),
step_type=StepType.GEOMETRY,
hash=sig,
)
def resolve_ast_target(target_step_ref: str, ast_registry: dict) -> str:
"""
V7.2: Looks up the actual math expression for an AST node ID.
The Planner works with IDs only โ€” this is where IDs are resolved to real math.
Raises KeyError if the reference is invalid, preventing silent hallucination.
"""
if target_step_ref not in ast_registry:
raise KeyError(
f"[RESOLVER] AST node '{target_step_ref}' not found in registry. "
f"Known nodes: {list(ast_registry.keys())}"
)
return ast_registry[target_step_ref]
# ==================== V7.2.1: DETERMINISTIC SOLVER DISPATCHER ====================
def execute_action(
action: str,
expression: str,
problem_id: str,
step_id: str,
ast_variables: list = None,
context: Optional[ActionContext] = None
) -> dict:
"""
V7.2.1 Wall 2: Deterministic Math Engine.
Routes a Planner Enum command to SymPy, executes it deterministically,
and returns a SHA256-signed step result.
Multi-variable handling (CTO note): if the expression has multiple free
symbols, we solve for the first variable listed in `ast_variables` (from
AST metadata). If no list is provided, we default to the first free symbol
found by SymPy โ€” never guess blindly.
Args:
action: One of ComputeAction enum values (string)
expression: Raw math expression string from AST registry
problem_id: For hash seeding
step_id: For hash seeding and tracking
ast_variables: Ordered list of variable names from build_ast_metadata()
Returns:
dict: {"id": step_id, "hash": "sha256...", "expression": "..."}
"""
import sympy as sp
# Parse expression โ€” treat '=' as LHS - RHS for sp.solve
normalized = expression.replace('=', '-')
try:
expr = sp.sympify(normalized, evaluate=False)
except Exception as e:
raise ValueError(f"[SOLVER] Cannot parse expression '{expression}': {e}")
# Multi-variable: determine which symbol to solve for
free_syms = list(expr.free_symbols)
solve_for = None
if action == "SOLVE_EQUATION":
if ast_variables:
# Use the first AST-declared variable that is actually in the expression
for var_name in ast_variables:
candidate = sp.Symbol(var_name)
if candidate in free_syms:
solve_for = candidate
break
if solve_for is None and free_syms:
# Fallback: first free symbol (alphabetically for determinism)
solve_for = sorted(free_syms, key=lambda s: s.name)[0]
logger.warning(
f"[SOLVER] No ast_variables hint provided. "
f"Solving for '{solve_for}' (first free symbol in expression)."
)
# โ”€โ”€ V7.3: Geometry Guards โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
if action == "CALCULATE_DISTANCE" and (
not context or not context.center or not context.point_a
):
raise ValueError(
"MissingContextError: CALCULATE_DISTANCE requires context.center and context.point_a"
)
if action == "CALCULATE_SLOPE_AND_LINE" and (
not context or not context.center or not context.point_a
):
raise ValueError(
"MissingContextError: CALCULATE_SLOPE_AND_LINE requires context.center and context.point_a"
)
# โ”€โ”€ V7.3: Geometry Handler Dispatch โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
if action == "FIND_AXIS_INTERSECTIONS":
# Substitute x=0 โ†’ solve for y, then y=0 โ†’ solve for x
x_sym, y_sym = sp.Symbol('x'), sp.Symbol('y')
y_intercepts = sp.solve(expr.subs(x_sym, 0), y_sym)
x_intercepts = sp.solve(expr.subs(y_sym, 0), x_sym)
# Build human-readable result list
result_parts = []
for yi in y_intercepts:
result_parts.append(f"(0, {yi})") # x=0
for xi in x_intercepts:
result_parts.append(f"({xi}, 0)") # y=0
result = result_parts if result_parts else ["ืื™ืŸ ื ืงื•ื“ื•ืช ื—ื™ืชื•ืš ืขื ื”ืฆื™ืจื™ื"]
logger.info(f"[SOLVER] โœ… FIND_AXIS_INTERSECTIONS on '{expression}' โ†’ {result}")
return sign_step_geometry(result, problem_id, step_id)
elif action == "CALCULATE_SLOPE_AND_LINE":
cx, cy = context.center # e.g. (3, 4)
px, py = context.point_a # e.g. (0, 0) โ€” origin
# Vertical line guard
if px == cx:
line_eq = f"x = {cx}"
result = [line_eq]
else:
slope = sp.Rational(cy - py, cx - px) # exact fraction (no float)
# y - py = slope*(x - px) โ†’ y = slope*x + b
b = py - slope * px
x_sym = sp.Symbol('x')
line_expr = slope * x_sym + b
# Simplify and return LaTeX-friendly string
result = [str(sp.simplify(line_expr))]
logger.info(f"[SOLVER] โœ… CALCULATE_SLOPE_AND_LINE center={context.center} โ†’ {result}")
return sign_step_geometry(result, problem_id, step_id)
elif action == "CALCULATE_DISTANCE":
cx, cy = context.center
px, py = context.point_a
# Use exact Rational arithmetic to avoid float comparison ambiguity
cx_r, cy_r = sp.Rational(cx).limit_denominator(1000), sp.Rational(cy).limit_denominator(1000)
px_r, py_r = sp.Rational(px).limit_denominator(1000), sp.Rational(py).limit_denominator(1000)
dist = sp.sqrt((px_r - cx_r)**2 + (py_r - cy_r)**2)
dist_val = sp.simplify(dist) # SymPy exact value (e.g. 5)
radius_sym = sp.Rational(context.radius).limit_denominator(1000)
# ๐Ÿšจ [SOP FIX] Guard against NoneType before float cast
for val in [dist_val, radius_sym]:
if val is None or val == "null" or val == "":
raise ValueError("Missing required mathematical argument from LLM. Cannot cast None to float.")
# Comparison via Python float (avoids SymPy Float vs int ambiguity)
dist_float = float(dist_val)
radius_float = float(radius_sym)
if abs(dist_float - radius_float) < 1e-9:
on_circle = "ืขืœ ื”ืžืขื’ืœ"
elif dist_float < radius_float:
on_circle = "ื‘ืชื•ืš ื”ืžืขื’ืœ"
else:
on_circle = "ืžื—ื•ืฅ ืœืžืขื’ืœ"
result = [f"d = {dist_val}", f"r = {context.radius}", on_circle]
logger.info(f"[SOLVER] โœ… CALCULATE_DISTANCE point={context.point_a} center={context.center} โ†’ {result}")
return sign_step_geometry(result, problem_id, step_id)
# โ”€โ”€ Legacy Algebraic Action Dispatch โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
ACTION_MAP = {
"SOLVE_EQUATION": lambda e: sp.solve(e, solve_for) if solve_for else sp.solve(e),
"SIMPLIFY": lambda e: [sp.simplify(e)],
"FACTOR": lambda e: [sp.factor(e)],
"EXPAND": lambda e: [sp.expand(e)],
"FIND_DERIVATIVE": lambda e: [sp.diff(e, solve_for or (free_syms[0] if free_syms else sp.Symbol('x')))],
"FIND_INTEGRAL": lambda e: [sp.integrate(e, solve_for or (free_syms[0] if free_syms else sp.Symbol('x')))],
"SUBSTITUTE": lambda e: [e],
}
fn = ACTION_MAP.get(action)
if not fn:
raise ValueError(f"[SOLVER] Unknown action: '{action}'. Check ComputeAction enum.")
try:
result = fn(expr)
if not isinstance(result, list):
result = [result]
logger.info(f"[SOLVER] โœ… {action} on '{expression}' โ†’ {result}")
except Exception as e:
raise RuntimeError(f"[SOLVER] SymPy failed on action '{action}': {e}")
return sign_step(result, problem_id, step_id)
class MathState(BaseModel):
"""ื™ื™ืฆื•ื’ ืžืชืžื˜ื™ ืื—ื™ื“ ืฉืœ ืžืฆื‘ ื”ื‘ืขื™ื”"""
equations: List[Any] = Field(default_factory=list) # ืจืฉื™ืžืช ืžืฉื•ื•ืื•ืช SymPy
solved_vars: Dict[Any, Any] = Field(default_factory=dict) # ืžืฉืชื ื™ื ืฉื ืคืชืจื•
original_text: str = ""
class Config:
arbitrary_types_allowed = True
def is_solved(self) -> bool:
# ื‘ืžืขืจื›ืช ืžืฉื•ื•ืื•ืช MVP: ืื ื™ืฉ ืžืฉืชื ื™ื ืคืชื•ืจื™ื ื•ืื™ืŸ ืขื•ื“ ืžืฉื•ื•ืื•ืช ื‘ืœืชื™ ืคืชื•ืจื•ืช ืจืœื•ื•ื ื˜ื™ื•ืช
if not self.equations:
return len(self.solved_vars) > 0
all_syms = set()
for eq in self.equations:
all_syms.update(eq.free_symbols)
# ื ืคืชืจ ืื ื›ืœ ื”ืžืฉืชื ื™ื ืžืงื‘ืœื™ื ืขืจืš
return len(all_syms) > 0 and all(sym in self.solved_vars for sym in all_syms)
# ==================== V5.8.0 RULE ENGINE MVP ====================
class MathRule:
name: str = "BaseRule"
def is_applicable(self, state: MathState) -> bool:
return False
def apply(self, state: MathState) -> Tuple[MathState, Any]:
raise NotImplementedError()
class RuleSolveLinearSystem(MathRule):
name = "ืคืชืจื•ืŸ ืžืขืจื›ืช ืžืฉื•ื•ืื•ืช"
def __init__(self):
self.x, self.y = symbols('x y')
def is_applicable(self, state: MathState) -> bool:
# ื ื–ื”ื” ืžืขืจื›ืช ืฉืœ 2 ืžืฉื•ื•ืื•ืช ืขื 2 ื ืขืœืžื™ื ื—ื•ืคืฉื™ื™ื
if len(state.equations) >= 2:
syms = set()
for eq in state.equations:
syms.update(eq.free_symbols)
if len(syms) == 2:
return True
return False
def apply(self, state: MathState) -> Tuple[MathState, Any]:
new_state = copy.copy(state)
# ื ื™ืกื™ื•ืŸ ืคืชืจื•ืŸ ืกื™ืžื‘ื•ืœื™ ืœืžืขืจื›ืช ื”ืžืฉื•ื•ืื•ืช ื™ื—ื“
eq1 = state.equations[0]
eq2 = state.equations[1]
# V5.8.1: Detailed Steps Mode (Step Granularity)
# Check if both equations are of the form "sym = expression" and have the same LHS
if eq1.lhs == eq2.lhs and isinstance(eq1.lhs, Symbol):
lhs_sym = eq1.lhs
# Get the other symbol playing the role of x
rhs_syms = list((eq1.rhs.free_symbols | eq2.rhs.free_symbols) - {lhs_sym})
if len(rhs_syms) == 1:
rhs_sym = rhs_syms[0]
steps = []
# 1. Equate
eq_step = Eq(eq1.rhs, eq2.rhs)
steps.append({"logic": "ื ืฉื•ื•ื” ื‘ื™ืŸ ืฉืชื™ ื”ืžืฉื•ื•ืื•ืช", "math": latex(eq_step), "rule_id": "solve_linear_system"})
# 2. Collect terms
diff_expr = eq_step.lhs - eq_step.rhs
c = diff_expr.coeff(rhs_sym)
d = diff_expr.subs(rhs_sym, 0)
collected_eq = Eq(c * rhs_sym, -d)
steps.append({"logic": "ื ืขื‘ื™ืจ ืื’ืคื™ื ื•ื ื›ื ืก ืื™ื‘ืจื™ื ื“ื•ืžื™ื", "math": latex(collected_eq), "rule_id": "solve_linear_system"})
# 3. Isolate variable
x_val = -d / c
isolated_eq = Eq(rhs_sym, x_val)
steps.append({"logic": f"ื ื—ืœืง ื‘ืžืงื“ื ืฉืœ {latex(rhs_sym)}", "math": latex(isolated_eq), "rule_id": "solve_linear_system"})
# 4. Substitute back
y_val = eq1.rhs.subs(rhs_sym, x_val)
# string replacement for substitution display to avoid auto-evaluation collapsing it
subs_str = latex(eq1.rhs).replace(latex(rhs_sym), f"({latex(x_val)})")
steps.append({"logic": f"ื ืฆื™ื‘ ืืช {latex(rhs_sym)} ื‘ืื—ืช ื”ืžืฉื•ื•ืื•ืช", "math": f"{latex(lhs_sym)} = {subs_str} = {latex(y_val)}", "rule_id": "solve_linear_system"})
new_state.solved_vars[rhs_sym] = x_val
new_state.solved_vars[lhs_sym] = y_val
new_state.equations = []
return new_state, steps
# V5.8.3 The Ultimate Guard: No step breakdown -> No solve
return state, "ืœื ื ื™ืชืŸ ืœืคืจืง ืœืฆืขื“ื™ื ืžื“ื•ืจื’ื™ื."
class RuleIsolateVariable(MathRule):
name = "ื‘ื™ื“ื•ื“ ืžืฉืชื ื”"
def is_applicable(self, state: MathState) -> bool:
# ื”ืื ื™ืฉ ืžืฉื•ื•ืื” ืื—ืช ืฉื ื™ืชืŸ ืœื‘ื•ื“ื“ ืžืžื ื” ืžืฉืชื ื” ืฉืขื“ื™ื™ืŸ ืœื ื‘ื•ื“ื“?
if len(state.equations) == 1:
return True
return False
def apply(self, state: MathState) -> Tuple[MathState, Any]:
new_state = copy.copy(state)
eq = state.equations[0]
syms = list(eq.free_symbols)
if not syms: return state, "ืื™ืŸ ืžืฉืชื ื™ื ืœื‘ื™ื“ื•ื“."
# ื ื ืกื” ืœื‘ื•ื“ื“ ืืช ื”ืžืฉืชื ื” (ืœืžืฉืœ x)
sol = solve(eq, syms[0])
if sol:
new_state.solved_vars[syms[0]] = sol[0]
new_state.equations = []
steps = [{"logic": "ื ื‘ื•ื“ื“ ืืช ื”ืžืฉืชื ื”", "math": f"{latex(syms[0])} = {latex(sol[0])}", "rule_id": "isolate_variable"}]
return new_state, steps
return state, "ืœื ื ื™ืชืŸ ืœื‘ื•ื“ื“ ืžืฉืชื ื”"
class RuleSubstitute(MathRule):
name = "ื”ืฆื‘ืช ืžืฉืชื ื” ืฉื‘ื•ื“ื“"
def is_applicable(self, state: MathState) -> bool:
return len(state.solved_vars) > 0 and len(state.equations) > 0
def apply(self, state: MathState) -> Tuple[MathState, Any]:
new_state = copy.copy(state)
# ื ืฆื™ื‘ ืืช ื›ืœ ื”ืžืฉืชื ื™ื ื”ื™ื“ื•ืขื™ื ื‘ืชื•ืš ื”ืžืฉื•ื•ืื•ืช ืฉื ื•ืชืจื•
new_eqs = []
for eq in state.equations:
new_eq = eq.subs(state.solved_vars)
new_eqs.append(new_eq)
new_state.equations = new_eqs
# ืื ื—ื ื• ืœื ืžื•ื—ืงื™ื ืืช state.solved_vars ื›ื™ ื ืจืฆื” ืœื–ื›ื•ืจ ืื•ืชื ืœื”ืžืฉืš
str_vars = ", ".join([f"{latex(k)} = {latex(v)}" for k,v in state.solved_vars.items()])
steps = [{"logic": "ื ืฆื™ื‘ ืืช ื”ืขืจื›ื™ื ื”ื™ื“ื•ืขื™ื ื‘ืžืฉื•ื•ืื•ืช", "math": str_vars, "rule_id": "substitute"}]
return new_state, steps
class StepValidator:
"""
V6.1 Phase 3: Validation Authority
Acts as the 'Supreme Court' for LLM proposed ProofGraphs.
"""
ALLOWED_CONSTANTS = {'pi', 'E', 'sqrt(2)'} # Removed 'I' as per QA Gate requirements.
@classmethod
def sympy_simplify_tolerance_guard(cls, expr_n, expr_n_plus_1) -> bool:
"""
V6.1 Phase 4: Tolerance Guard
Numerical evaluation fallback to handle micro-variances that avoid simplification.
"""
import random
try:
diff_expr = expr_n - expr_n_plus_1
free_syms = diff_expr.free_symbols
# If no free symbols, just evaluate numerically
if not free_syms:
val = diff_expr.evalf()
# ๐Ÿšจ [SOP FIX] Guard against NoneType before float cast
if val is None or val == "null" or val == "":
raise ValueError("Missing required mathematical argument from LLM. Cannot cast None to float.")
return abs(float(val)) < 1e-9
# Sampling: 10 random points
for _ in range(10):
subs = {s: random.uniform(0.1, 10.0) for s in free_syms}
val = diff_expr.evalf(subs=subs)
# ๐Ÿšจ [SOP FIX] Guard against NoneType before float cast
if val is None or val == "null" or val == "":
raise ValueError("Missing required mathematical argument from LLM. Cannot cast None to float.")
if abs(float(val)) > 1e-9: # Precision set to 10^-9 as per QA Gate
return False
return True
except Exception:
return False
@classmethod
def validate_transition(cls, step_n: str, step_n_plus_1: str) -> bool:
"""
Deterministically verifies the algebraic equivalence between two steps.
"""
try:
expr_n = sympify(str(step_n).replace('=', '-'), evaluate=False)
expr_n_plus_1 = sympify(str(step_n_plus_1).replace('=', '-'), evaluate=False)
# Use simplify to check equivalence
diff = simplify(expr_n - expr_n_plus_1)
if diff == 0:
return True
# Fallback for trigonometric identities or complex simplification
if trigsimp(expr_n) == trigsimp(expr_n_plus_1):
return True
# V6.1 Phase 4: Tolerance Guard (Secondary Check)
if cls.sympy_simplify_tolerance_guard(expr_n, expr_n_plus_1):
logger.info(f"๐Ÿ›ก๏ธ [VALIDATOR] Tolerance Guard PASSED for {step_n} -> {step_n_plus_1}")
return True
# If equivalence fails, it might be an irreversible action (e.g. squaring).
return False
except Exception as e:
logger.warning(f"[VALIDATOR] Transition validation error '{step_n}' -> '{step_n_plus_1}': {e}")
return False
@classmethod
def check_proofgraph_closure(cls, initial_math: str, final_step: str) -> bool:
"""
Ensures all variables in the original problem are resolved or accounted for,
and no hallucinations occurred via injected fake constants/variables.
"""
try:
initial_exprs = [sympify(str(p).replace('=', '-'), evaluate=False) for p in str(initial_math).split(',')]
initial_vars = set()
for ex in initial_exprs:
initial_vars.update(ex.free_symbols)
final_exprs = [sympify(str(p).replace('=', '-'), evaluate=False) for p in str(final_step).split(',')]
final_vars = set()
for ex in final_exprs:
final_vars.update(ex.free_symbols)
filtered_final_vars = {str(v) for v in final_vars if not v.is_number and str(v) not in cls.ALLOWED_CONSTANTS}
filtered_initial_vars = {str(v) for v in initial_vars if not v.is_number and str(v) not in cls.ALLOWED_CONSTANTS}
if len(filtered_final_vars - filtered_initial_vars) > 0:
logger.warning(f"[VALIDATOR] Closure check failed. Leaked vars: {filtered_final_vars - filtered_initial_vars}")
return False
return True
except Exception as e:
logger.warning(f"[VALIDATOR] Closure validation error: {e}")
return False
@classmethod
def evaluate_proposal(cls, initial_math: str, draft_steps: list) -> tuple[float, str]:
"""
Evaluates the entire Draft ProofGraph.
Returns (Validation Score [0.0 - 1.0], Error Reason)
"""
if not draft_steps:
return 0.0, "EMPTY_DRAFT"
valid_transitions = 0
total_transitions = len(draft_steps)
# We assume step 0 is the initial math or a reformatted version of it.
# We validate transition from step[i] to step[i+1]
for i in range(len(draft_steps) - 1):
math_n = draft_steps[i].get('math', '')
math_n_plus_1 = draft_steps[i+1].get('math', '')
if cls.validate_transition(math_n, math_n_plus_1):
valid_transitions += 1
else:
logger.warning(f"[VALIDATOR] Rejecting step {i+1} -> {i+2}: {math_n} to {math_n_plus_1}")
return 0.0, f"ืžืขื‘ืจ ืžืชืžื˜ื™ ืฉื’ื•ื™ ื‘ื™ืŸ: {math_n} ืœื‘ื™ืŸ {math_n_plus_1}"
score = valid_transitions / total_transitions if total_transitions > 0 else 0.0
# Check Closure
final_step_math = draft_steps[-1].get('math', '')
if not cls.check_proofgraph_closure(initial_math, final_step_math):
return 0.0, "ืกื’ื™ืจืช ืคืชืจื•ืŸ ืœื ื—ื•ืงื™ืช (ื ืžืฆืื• ืžืฉืชื ื™ื ืœื ืžืื•ืฉืจื™ื)"
return score, ""
def __init__(self, success=False, function=None, derivative=None, steps=None, operator_used=None):
self.success = success
self.function = function
self.derivative = derivative
self.steps = steps or [] # ื—ื•ื‘ื” ืขื‘ื•ืจ ื”-ProofGraph
self.operator_used = operator_used
class SmartSolver:
def __init__(self):
print("โœ… ๐ŸŸข [BIT-LOG: SmartSolver V263.0] - Algebra Engine Active")
def solve(self, context):
math_input = context.math_input
category = context.category
original_text = getattr(context, 'original_text', "").lower()
grade_num = getattr(context, 'grade_num', 12)
print(f"๐Ÿ” [BIT-LOG: SOLVER] Analyzing input: '{math_input}'")
print(f"๐Ÿ” [BIT-LOG: SOLVER] Category: {category}, Intent Text: '{original_text[:50]}...'")
# --- V5.8.0 Rule Engine MVP (Always First) ---
print("๐Ÿ”ข [BIT-LOG: SOLVER] Triggering Rule Engine MVP")
rule_engine_res = self._solve_linear_system(math_input)
if rule_engine_res and rule_engine_res.success:
return rule_engine_res
# --- ืขื ืฃ 2: ื—ืงื™ืจืช ืคื•ื ืงืฆื™ื•ืช (ื›ื™ืชื” ื™' ืขื“ ื™"ื‘) ---
# V4.2.12: Anchor regex to start of string to avoid matching 4x + 5y = 120 as 'y = 120'
is_explicit_func = bool(re.match(r'^(f\(x\)|y|g\(x\))\s*=', math_input, re.IGNORECASE))
investigation_keywords = ["ื—ืงื•ืจ", "ื—ืงื™ืจืช", "ืงื™ืฆื•ืŸ", "ืืกื™ืžืคื˜ื•ื˜", "ื ื’ื–ืจืช", "ืขืœื™ื”", "ื™ืจื™ื“ื”"]
has_intent = any(kw in original_text for kw in investigation_keywords)
# ื”ื ื’ื–ืจืช ืชื•ืคืขืœ ืจืง ืื: ื–ื• ื—ืงื™ืจื” + ื™ืฉ ืคื•ื ืงืฆื™ื” ืžืคื•ืจืฉืช + ื™ืฉ ื›ื•ื•ื ื” ื‘ื˜ืงืกื˜
should_derive = (category == "INVESTIGATION" and is_explicit_func and has_intent and grade_num > 7)
if should_derive:
print("๐Ÿ“ˆ [BIT-LOG: SOLVER] Triggering Calculus Engine (Derivatives)")
return self._solve_calculus(math_input)
# Fallback ื’ื ืจื™
print(f"๐Ÿ›ก๏ธ [BIT-LOG: SOLVER] Generic Algebra Mode. category={category}, intent={has_intent}")
return SmartResult(success=True, function=math_input, operator_used="ALGEBRA_GENERAL")
def _solve_linear_system(self, raw_input):
"""V5.8.0: ืžื ื•ืข ืงื‘ืœืช ื”ื—ืœื˜ื•ืช ืžื‘ื•ืกืก MVP ื—ื•ืงื™ื"""
try:
# ืฉืœื‘ 1: Parsing
eq_parts = re.split(r'[,\n]', raw_input)
parsed_eqs = []
for part in eq_parts:
if '=' in part:
s_a, s_b = part.split('=')
parsed_eqs.append(Eq(sympify(self._sanitize(s_a)), sympify(self._sanitize(s_b))))
if not parsed_eqs:
return SmartResult(success=False)
current_state = MathState(equations=parsed_eqs, original_text=raw_input)
rules = [RuleSolveLinearSystem(), RuleIsolateVariable(), RuleSubstitute()]
proof_graph_steps = []
iteration = 0
MAX_ITER = 5
logger.info(f"๐Ÿงฎ [RULE-ENGINE] Starting resolution for system: {parsed_eqs}")
# ืฉืœื‘ 2: ืœื•ืœืืช ื”ื—ื•ืงื™ื ื”ืืจื›ื™ื˜ืงื˜ื•ื ื™ืช (The Engine)
while not current_state.is_solved() and iteration < MAX_ITER:
applied_any = False
for rule in rules:
if rule.is_applicable(current_state):
new_state, step_data = rule.apply(current_state)
if isinstance(step_data, list):
for s in step_data:
proof_graph_steps.append({
"id": len(proof_graph_steps) + 1,
"logic": s["logic"],
"math": s["math"]
})
else:
proof_graph_steps.append({
"id": len(proof_graph_steps) + 1,
"logic": rule.name,
"math": step_data
})
current_state = new_state
applied_any = True
break # Start rules over with new state
if not applied_any:
logger.warning("๐Ÿงฎ [RULE-ENGINE] No applicable rules found to continue solving.")
break
iteration += 1
# ื™ืฆื™ืจืช ืžื•ื“ืœ SmartResult ืขื ื”ื•ื›ื—ื” ืžื‘ื•ืกืกืช ื—ื•ืงื™ื
if current_state.is_solved():
print(f"โœ… [RULE-ENGINE] Resolved system to state: {current_state.solved_vars}")
# ื”ื•ืกืคืช ืฆืขื“ ืกื™ื›ื•ื ืื—ืจื•ืŸ
res_str = ", ".join([f"{latex(k)}={latex(v)}" for k,v in current_state.solved_vars.items()])
return SmartResult(success=True, function=raw_input, steps=proof_graph_steps, operator_used="RULE_ENGINE_SYSTEM")
else:
return SmartResult(success=False)
except Exception as e:
print(f"โŒ [BIT-LOG: SOLVER] Algebra Rule Engine Error: {e}")
return SmartResult(success=False)
def _solve_calculus(self, math_input):
try:
x = symbols('x')
# ื—ื™ืœื•ืฅ ื”ื‘ื™ื˜ื•ื™ ืื—ืจื™ ื”- '='
expr_part = math_input.split('=')[-1]
expr_str = self._sanitize(expr_part)
logger.info(f"๐Ÿงฎ [TRACE] EXPRESSION SENT TO SYMPY: {expr_str}")
expr = sympify(expr_str)
f_prime = diff(expr, x)
# ื—ื™ืฉื•ื‘ ื ืงื•ื“ื•ืช ืงื™ืฆื•ืŸ
crit_points = []
try:
sols = solve(f_prime, x)
for s in sols:
if s is not None and s.is_real:
y_val = expr.subs(x, s).evalf()
if y_val is not None:
crit_points.append(f"({float(s):.2f}, {float(y_val):.2f})")
except Exception as e:
logger.warning(f"[SOLVER] Calculus point evaluation failed: {e}")
steps = [{"id": 1, "math": f"f'(x)={latex(f_prime)}", "logic": "ื—ื™ืฉื•ื‘ ื ื’ื–ืจืช"}]
return SmartResult(
success=True,
function=f"f(x)={latex(expr)}",
derivative=f"f'(x)={latex(f_prime)}",
steps=steps,
operator_used="DERIVATIVE"
)
except Exception as e:
print(f"โŒ [BIT-LOG: SOLVER] Calculus Engine Error: {e}")
return SmartResult(success=False)
def _sanitize(self, text):
"""โœ… ืกื ื›ืจื•ืŸ ืขื ืžื ื’ื ื•ืŸ ื”ื ื™ืงื•ื™ ื”ืžืจื›ื–ื™ (V260.0)"""
text = text.replace(r'\left', '').replace(r'\right', '')
while r'\frac' in text:
text = re.sub(r'\\frac\s*\{(.*?)\}\{(.*?)\}', r'(\1)/(\2)', text)
text = re.sub(r'(sin|cos|tan)\s+([a-zA-Z0-9]+)', r'\1(\2)', text)
text = text.replace(r'\sin', 'sin').replace(r'\cos', 'cos').replace(r'\tan', 'tan')
text = text.replace('^', '**')
# โœ… V260.0: Robust implicit multiplication
text = re.sub(r'(\d)([a-zA-Z(])', r'\1*\2', text)
text = re.sub(r'\)([a-zA-Z0-9(])', r')*\1', text)
text = re.sub(r'(?<![a-zA-Z])([a-zA-Z])\(', r'\1*(', text)
allowed = "0123456789+-*/().=xsincoabslpqrtABSpi eylog"
return ''.join(c for c in text if c.lower() in allowed).strip()