New tool: Magmas CLI

Stream: Equational

[Michael Bucko]

python3 magmas.py operation.txt equation_list.txt > solution.txt
https://github.com/teorth/equational_theories/pull/764

[Pietro Monticone]

Thanks @Michael Bucko. Could you please follow the workflow so that reviewers / maintainers know the full context ?

[Michael Bucko]

@Pietro Monticone Could you please link the workflow ticket? Gonna check it out.

[Michael Bucko]

Ah, thank you. I see your comment.

[Michael Bucko]

@Shreyas Srinivas Here's the related GitHub issue:
https://github.com/teorth/equational_theories/issues/802

[Shreyas Srinivas]

Ah so this was already discussed

[Shreyas Srinivas]

Let's see what others have to say

[Michael Bucko]

Yeah, it's a small CLI tool for people who use Python to build apps around operation tables. I've been using the FMT a lot, and needed this. So I thought I'll share it with others.

[Michael Bucko]

@Shreyas Srinivas I'd appreciate your feedback here too :thank_you:

[Shreyas Srinivas]

I have two questions:

1. Why do you want the following: "A Python function + CLI that takes the equation list as input, an operation table as an argument, and returns JSON with 1/0 for each equation."
2. Why can't it be generated from the info provided by lake exe extract_implications and the API that the equation_explorer uses?

[Shreyas Srinivas]

for 1: what does 1 or 0 mean?

[Shreyas Srinivas]

Reading the PR: you want to create an opt-able set but this is trivial to extract from the generator script for equations isn't it?

[Michael Bucko]

Shreyas Srinivas schrieb:

for 1: what does 1 or 0 mean?

1 - means satisfies, 0 - refutes. It's basically like the FMT but for Python developers, and CLI. (My use case: I wanted to use the function in an RL agent)

[Michael Bucko]

Shreyas Srinivas schrieb:

Reading the PR: you want to create an opt-able set but this is trivial to extract from the generator script for equations isn't it?

I basically needed a Python script for the RL work. My alternative was the FMT tool.

[Shreyas Srinivas]

But equations are not in and of themselves satisfied or refuted?

[Shreyas Srinivas]

I assume there is more context here

[Shreyas Srinivas]

I think building off the FMT tool is a good idea

[Shreyas Srinivas]

In that, you can simply run and output the implication list, process it, and see what RL yields

[Michael Bucko]

Shreyas Srinivas schrieb:

I have two questions:

1. Why do you want the following: "A Python function + CLI that takes the equation list as input, an operation table as an argument, and returns JSON with 1/0 for each equation."
2. Why can't it be generated from the info provided by lake exe extract_implications and the API that the equation_explorer uses?

1. I need something like the FMT for Python developers.
2. I use extract_implications too, but that takes a bit. The use case is closer to the FMT tool.

[Michael Bucko]

Shreyas Srinivas schrieb:

I think building off the FMT tool is a good idea

Yes, it's like FMT for Python developers who want to write scripts around optables and perhaps RL (I wrote a simple RL agent as part of my experiments).

[Shreyas Srinivas]

but surely python can read the output format of the FMT tool and process it as a dictionary

[Michael Bucko]

Shreyas Srinivas schrieb:

but surely python can read the output format of the FMT tool and process it as a dictionary

Yes, it could (assuming that the FMT has an API; I don't know about it). But SW-wise, it's not a bad idea to have a Python version.

[Shreyas Srinivas]

okay, so how would this work?

[Shreyas Srinivas]

In your PR I see another copy of the equation list

[Shreyas Srinivas]

And the text data included

[Michael Bucko]

You run it like this

python3 magmas.py operation.txt equation_list.txt > solution.txt

You can use the function from the code in Python based RL research.

[Shreyas Srinivas]

I mean they should be gitignored

[Shreyas Srinivas]

Secondly I see you are generating the equation_list. I would prefer to see the existing equation generation scripts to be used

[Shreyas Srinivas]

In fact it's better if your initial data structure comes from the FME tool. Although I am still not sure how much of a difference the script in and of itself makes

[Shreyas Srinivas]

Can you combine it with the PR where you are using its output to do something like using RL for counterexamples?

[Shreyas Srinivas]

Maybe the developer of the FME tool can pitch in

[Shreyas Srinivas]

(deleted)

[Michael Bucko]

Tbh I have certain time constraints atm (kids :joy: ), so please just decide whether you'd need the Python version.

(Yes, the equations file, as you mentioned, is not in .gitignore, because it's slightly different, i.e. it helps me to see the equation numbers in front of the equations).

When it comes to the RL experiments, I'll probably first make a separate post.

[Shreyas Srinivas]

My take is, it would make sense to combine this PR with the results of the RL experiments formalised. Then we'd have something concrete to go on.

[Michael Bucko]

The rl agent is still being an ongoing thing. It works (it's still not fully done, more like 90% done, look at episode number, the reward setup, and the 854n1055) like this:

(menv) riccitensor@perelman equational_theories % python3 rlmagma.py

Starting magma implication explorer...

Starting training with magma size 3...

Looking for counterexample where:

  1. Equation 854 is SATISFIED

  2. Equation 1055 is NOT SATISFIED

Episode 1/1000

Steps in this episode: 64544

Total actions taken: 64544

Unique states explored: 7532

Time elapsed: 5.0 seconds

[Michael Bucko]

But it still has issues / I still need to fix a couple of things. I'd need to work on it. My case was 854.

[Shreyas Srinivas]

Then let's take the time and get results from this before pushing ahead with the PR

[Shreyas Srinivas]

(btw could you mark it as a draft please?)

[Michael Bucko]

Sure! I'll make it a draft.

[Michael Bucko]

It's a draft now.

[Shreyas Srinivas]

Thanks :)

[Michael Bucko]

btw. the rl thing is still to be fixed (but it's 90% there), but I can share here so that you can see how I am using these functions

import numpy as np
import re
from itertools import product
import time

class MagmaEnvironment:
    def __init__(self, size=3):
        self.size = size
        self.current_operation_table = np.random.randint(0, self.size, (self.size, self.size))
        self.steps_taken = 0

    def shunting_yard(self, expression):
        output, stack = [], []
        tokens = re.findall(r'[a-z]+|[◇*()]', expression)
        precedence = {'◇': 1, '*': 1}

        for token in tokens:
            if token.isalpha():
                output.append(token)
            elif token in precedence:
                while (stack and stack[-1] in precedence and
                       precedence[token] <= precedence[stack[-1]]):
                    output.append(stack.pop())
                stack.append(token)
            elif token == '(':
                stack.append(token)
            elif token == ')':
                while stack and stack[-1] != '(':
                    output.append(stack.pop())
                if stack:
                    stack.pop()

        while stack:
            if stack[-1] != '(':
                output.append(stack.pop())
            else:
                stack.pop()

        return output

    def evaluate_rpn(self, rpn, op_table, variables):
        stack = []
        for token in rpn:
            if token.isalpha():
                if token not in variables:
                    return None
                stack.append(variables[token])
            elif token == '◇':
                if len(stack) < 2:
                    return None
                b, a = stack.pop(), stack.pop()
                stack.append(op_table[a][b])
        return stack[0] if stack else None

    def get_variables(self, expression):
        return set(re.findall(r'[a-z]', expression))

    def check_equation_satisfaction(self, equation, op_table):
        try:
            lhs_expr, rhs_expr = equation.split('=')
            lhs_rpn = self.shunting_yard(lhs_expr.strip())
            rhs_rpn = self.shunting_yard(rhs_expr.strip())

            variables = self.get_variables(equation)
            table_size = len(op_table)

            value_combinations = product(range(table_size), repeat=len(variables))

            for values in value_combinations:
                var_assignment = dict(zip(sorted(variables), values))

                lhs_result = self.evaluate_rpn(lhs_rpn, op_table, var_assignment)
                rhs_result = self.evaluate_rpn(rhs_rpn, op_table, var_assignment)

                if lhs_result is None or rhs_result is None:
                    continue

                if lhs_result != rhs_result:
                    return False, var_assignment

            return True, None

        except Exception as e:
            print(f"Error processing equation: {equation}")
            print(f"Error details: {str(e)}")
            return False, None

    def check_implication(self, operation_table):
        equation_854 = "x = x ◇ ((y ◇ z) ◇ (x ◇ z))"
        equation_1055 = "x = x ◇ ((y ◇ (z ◇ x)) ◇ x)"

        eq854_satisfied, _ = self.check_equation_satisfaction(equation_854, operation_table)
        eq1055_satisfied, counterexample = self.check_equation_satisfaction(equation_1055, operation_table)

        if eq854_satisfied and not eq1055_satisfied:
            return True, counterexample
        else:
            return False, None

    def step(self, action):
        self.steps_taken += 1
        i, j = action
        self.current_operation_table[i][j] = np.random.randint(0, self.size)

        found_counterexample, counterexample = self.check_implication(self.current_operation_table)

        if found_counterexample:
            reward = 1.0
            done = True
        else:
            reward = 0.0
            done = False

        state = self.get_state()
        return state, reward, done, {
            "counterexample": counterexample,
            "steps": self.steps_taken
        }

    def get_state(self):
        return self.current_operation_table.copy()

    def reset(self):
        self.steps_taken = 0
        self.current_operation_table = np.random.randint(0, self.size, (self.size, self.size))
        return self.get_state()

class QLearningAgent:
    def __init__(self, state_space_size, action_space_size, learning_rate=0.1, discount_factor=0.95,
epsilon=0.1):
        self.q_table = {}
        self.lr = learning_rate
        self.gamma = discount_factor
        self.epsilon = epsilon
        self.action_space_size = action_space_size
        self.total_actions = 0

    def get_action(self, state):
        self.total_actions += 1
        state_key = str(state.tobytes())

        if state_key not in self.q_table:
            self.q_table[state_key] = np.zeros(self.action_space_size)

        if np.random.random() < self.epsilon:
            return np.random.randint(0, self.action_space_size)
        else:
            return np.argmax(self.q_table[state_key])

    def update(self, state, action, reward, next_state):
        state_key = str(state.tobytes())
        next_state_key = str(next_state.tobytes())

        if next_state_key not in self.q_table:
            self.q_table[next_state_key] = np.zeros(self.action_space_size)

        old_value = self.q_table[state_key][action]
        next_max = np.max(self.q_table[next_state_key])

        new_value = (1 - self.lr) * old_value + self.lr * (reward + self.gamma * next_max)
        self.q_table[state_key][action] = new_value

def train_agent(episodes=1000, size=3, verbose=True):
    env = MagmaEnvironment(size=size)
    agent = QLearningAgent(size**2, size**2)

    start_time = time.time()
    last_update = start_time

    print(f"\nStarting training with magma size {size}...")
    print(f"Looking for counterexample where:")
    print(f"  1. Equation 854 is SATISFIED")
    print(f"  2. Equation 1055 is NOT SATISFIED")

    for episode in range(episodes):
        state = env.reset()
        done = False

        while not done:
            action = agent.get_action(state)
            action_tuple = (action // size, action % size)
            next_state, reward, done, info = env.step(action_tuple)

            agent.update(state, action, reward, next_state)
            state = next_state

            current_time = time.time()
            if verbose and current_time - last_update >= 5:
                print(f"\nEpisode {episode + 1}/{episodes}")
                print(f"Steps in this episode: {info['steps']}")
                print(f"Total actions taken: {agent.total_actions}")
                print(f"Unique states explored: {len(agent.q_table)}")
                print(f"Time elapsed: {current_time - start_time:.1f} seconds")
                last_update = current_time

            if done and info["counterexample"] is not None:
                print("\n🎉 Found valid counterexample!")
                print(f"\nOperation table:")
                print(env.current_operation_table)

                print(f"\nCounterexample variable assignment:")
                print(info["counterexample"])

                print(f"\nSearch statistics:")
                print(f"Episodes needed: {episode + 1}")
                print(f"Total steps taken: {agent.total_actions}")
                print(f"Unique states explored: {len(agent.q_table)}")
                print(f"Time taken: {time.time() - start_time:.1f} seconds")

                return info["counterexample"], env.current_operation_table

    print("\nNo counterexample found after all episodes.")
    print(f"Total steps taken: {agent.total_actions}")
    print(f"Unique states explored: {len(agent.q_table)}")
    print(f"Time taken: {time.time() - start_time:.1f} seconds")

    return None, None

if __name__ == "__main__":
    print("Starting the magma rl tool...")
    counterexample, operation_table = train_agent(episodes=1000, size=3, verbose=True)

[Michael Bucko]

I was testing on the 854N1055, but it still fails. The main thing is the Q-learning agent. The reward, I guess, is also not good atm. I was thinking of making a gymnasium env, too.

[Vlad Tsyrklevich]

Ah I just saw this thread so I had commented on the issue, but I'll reproduce it here as there is more discussion here:
Equations should really uses the JSON format from lake exe export_equations.
Also, isn't this tool already done by scripts/explore_magma.py ?

[Michael Bucko]

@Vlad Tsyrklevich I haven'T worked with explore_magma, so perhaps you could judge?

[Vlad Tsyrklevich]

python scripts/explore_magma.py "[[1, 1], [1, 1]]"

[Michael Bucko]

Wow, it works. My tool does not support this format and it's mostly for the RL functions. But I guess I could've checked that tool first. Thank you @Vlad Tsyrklevich

[Vlad Tsyrklevich]

If you want JSON it would probably not be too hard to add --json to match your output format

[Michael Bucko]

Yeah, trivial ofc. But the goal for me was the RL tool above, and quick tests with operation tables. I just didn't know about explore_magma.

[Zoltan A. Kocsis (Z.A.K.)]

@Shreyas Srinivas writes:

Maybe the developer of the FME tool can pitch in

That's me, I guess. Sorry, I'm busy assessing masters theses and preparing for AAL Conf 2024 at the moment. Until the latter concludes (28 Nov) I won't have time to assist with the project. But if there are things you or anyone else would like me to assist with, the best way to alert me is by creating Github issues: I'll attend to them next month if nobody else claims them in the meantime.

[Shreyas Srinivas]

Hey, sorry I forgot for a moment there that you were busy

[Shreyas Srinivas]

Apologies