Buckets:

tahamajs
/

Sysmem2_in_AI

Files

xet

tahamajs/Sysmem2_in_AI / ComputerAssignments /CA17_reinforcement_learning_math /scripts /demo.py

tahamajs

4 months ago

download

raw

9.02 kB

#!/usr/bin/env python3

import argparse
import os
import sys
import time
from pathlib import Path


sys.path.append(str(Path(__file__).parent.parent / "src"))

from src.agents.math_agent import MathReasoningAgent
from src.environments.math_environment import MathEnvironment
from src.config import Config
from src.utils.logging_utils import setup_logging
from loguru import logger

def parse_args():
    parser = argparse.ArgumentParser(description="Demo RL-LLM system")

    parser.add_argument(
        "--config", type=str, default="config.yaml", help="Path to configuration file"
    )

    parser.add_argument(
        "--model_name", type=str, default="gpt2", help="Model name to use"
    )

    parser.add_argument(
        "--num_problems", type=int, default=5, help="Number of problems to solve"
    )

    parser.add_argument(
        "--difficulty",
        type=str,
        default="medium",
        choices=["easy", "medium", "hard"],
        help="Difficulty level",
    )

    parser.add_argument(
        "--problem_type",
        type=str,
        default="arithmetic",
        choices=["arithmetic", "algebra", "geometry", "calculus"],
        help="Problem type",
    )

    parser.add_argument(
        "--device", type=str, default="auto", help="Device to use (cuda, cpu, auto)"
    )

    parser.add_argument("--interactive", action="store_true", help="Interactive mode")

    parser.add_argument(
        "--show_solutions", action="store_true", help="Show detailed solutions"
    )

    parser.add_argument(
        "--log_level",
        type=str,
        default="INFO",
        choices=["DEBUG", "INFO", "WARNING", "ERROR"],
        help="Logging level",
    )

    return parser.parse_args()

def create_sample_problems():
    problems = [
        {
            "problem": "Solve: 2x + 5 = 13",
            "type": "algebra",
            "difficulty": "easy",
            "expected_answer": "x = 4",
        },
        {
            "problem": "What is the area of a circle with radius 5?",
            "type": "geometry",
            "difficulty": "medium",
            "expected_answer": "78.54 (approximately)",
        },
        {
            "problem": "Calculate: 15 × 23 + 7 × 8",
            "type": "arithmetic",
            "difficulty": "easy",
            "expected_answer": "409",
        },
        {
            "problem": "Find the derivative of x² + 3x + 2",
            "type": "calculus",
            "difficulty": "medium",
            "expected_answer": "2x + 3",
        },
        {
            "problem": "Solve the quadratic equation: x² - 5x + 6 = 0",
            "type": "algebra",
            "difficulty": "medium",
            "expected_answer": "x = 2 or x = 3",
        },
    ]
    return problems

def run_demo_problem(
    agent: MathReasoningAgent,
    environment: MathEnvironment,
    problem_info: dict,
    show_solution: bool = False,
):
    problem = problem_info["problem"]
    problem_type = problem_info["type"]
    difficulty = problem_info["difficulty"]
    expected = problem_info.get("expected_answer", "Unknown")

    print(f"\n{'='*60}")
    print(f"Problem: {problem}")
    print(f"Type: {problem_type}, Difficulty: {difficulty}")
    print(f"Expected Answer: {expected}")
    print(f"{'='*60}")


    start_time = time.time()
    solution = agent.generate_solution(problem)
    generation_time = time.time() - start_time

    print(f"\nGenerated Solution:")
    print(f"{solution}")

    if show_solution:
        print(f"\nDetailed Analysis:")
        print(f"Generation time: {generation_time:.2f}s")
        print(f"Solution length: {len(solution)} characters")


    reward = environment.verify_solution(problem, solution)
    success = reward > 0

    print(f"\nVerification:")
    print(f"Reward: {reward:.2f}")
    print(f"Success: {'✅' if success else '❌'}")

    return {
        "problem": problem,
        "solution": solution,
        "reward": reward,
        "success": success,
        "generation_time": generation_time,
        "expected": expected,
    }

def interactive_mode(agent: MathReasoningAgent, environment: MathEnvironment):
    print("\n" + "=" * 60)
    print("INTERACTIVE MODE")
    print("=" * 60)
    print("Enter math problems to solve. Type 'quit' to exit.")
    print("Available problem types: arithmetic, algebra, geometry, calculus")
    print("Available difficulties: easy, medium, hard")
    print("=" * 60)

    while True:
        try:
            print("\n" + "-" * 40)
            problem = input("Enter a math problem: ").strip()

            if problem.lower() in ["quit", "exit", "q"]:
                print("Goodbye!")
                break

            if not problem:
                continue


            print(f"\nSolving: {problem}")
            start_time = time.time()
            solution = agent.generate_solution(problem)
            generation_time = time.time() - start_time

            print(f"\nSolution:")
            print(f"{solution}")


            reward = environment.verify_solution(problem, solution)
            success = reward > 0

            print(f"\nVerification:")
            print(f"Reward: {reward:.2f}")
            print(f"Success: {'✅' if success else '❌'}")
            print(f"Time: {generation_time:.2f}s")

        except KeyboardInterrupt:
            print("\nGoodbye!")
            break
        except Exception as e:
            print(f"Error: {e}")

def main():
    args = parse_args()


    log_file = "demo.log"
    setup_logging(log_file, args.log_level)

    logger.info("Starting RL-LLM demo")
    logger.info(f"Arguments: {args}")

    try:

        if os.path.exists(args.config):
            config = Config.from_yaml(args.config)
            logger.info(f"Loaded configuration from {args.config}")
        else:
            logger.warning(
                f"Configuration file {args.config} not found, using defaults"
            )
            config = Config()


        config.model.model_name = args.model_name
        config.model.device = args.device


        logger.info("Initializing agent and environment")
        agent = MathReasoningAgent(config.model)
        environment = MathEnvironment(config.environment)

        print("=" * 60)
        print("RL-LLM MATH REASONING DEMO")
        print("=" * 60)
        print(f"Model: {args.model_name}")
        print(f"Device: {config.model.device}")
        print(
            f"Agent initialized with {len(agent.few_shot_examples)} few-shot examples"
        )

        if args.interactive:
            interactive_mode(agent, environment)
        else:

            if args.num_problems > 0:
                print(f"\nRunning {args.num_problems} demo problems...")


                problems = []
                for i in range(args.num_problems):
                    problem = environment.generate_problem(
                        args.difficulty, args.problem_type
                    )
                    problems.append(
                        {
                            "problem": problem,
                            "type": args.problem_type,
                            "difficulty": args.difficulty,
                        }
                    )


                results = []
                for i, problem_info in enumerate(problems):
                    print(f"\nProblem {i+1}/{args.num_problems}")
                    result = run_demo_problem(
                        agent, environment, problem_info, args.show_solutions
                    )
                    results.append(result)


                print(f"\n{'='*60}")
                print("DEMO SUMMARY")
                print(f"{'='*60}")
                print(f"Total problems: {len(results)}")
                print(f"Successful: {sum(1 for r in results if r['success'])}")
                print(
                    f"Success rate: {sum(1 for r in results if r['success']) / len(results):.2%}"
                )
                print(
                    f"Average reward: {sum(r['reward'] for r in results) / len(results):.2f}"
                )
                print(
                    f"Average time: {sum(r['generation_time'] for r in results) / len(results):.2f}s"
                )


                print(f"\n{'='*60}")
                print("SAMPLE PROBLEMS")
                print(f"{'='*60}")
                sample_problems = create_sample_problems()

                for i, problem_info in enumerate(sample_problems[:3]):
                    print(f"\nSample {i+1}:")
                    print(f"Problem: {problem_info['problem']}")
                    print(
                        f"Type: {problem_info['type']}, Difficulty: {problem_info['difficulty']}"
                    )
                    print(f"Expected: {problem_info['expected_answer']}")

        logger.info("Demo completed successfully")

    except Exception as e:
        logger.error(f"Demo failed: {e}")
        print(f"\nError: {e}")
        raise

if __name__ == "__main__":
    main()

Xet Storage Details

Size:: 9.02 kB
Xet hash:: 4195fefc91d0eaf29ebc3e363295832e7f64bb9d4dfb2d323a1ece4c5830563a

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	#!/usr/bin/env python3

	import argparse
	import os
	import sys
	import time
	from pathlib import Path


	sys.path.append(str(Path(__file__).parent.parent / "src"))

	from src.agents.math_agent import MathReasoningAgent
	from src.environments.math_environment import MathEnvironment
	from src.config import Config
	from src.utils.logging_utils import setup_logging
	from loguru import logger

	def parse_args():
	parser = argparse.ArgumentParser(description="Demo RL-LLM system")

	parser.add_argument(
	"--config", type=str, default="config.yaml", help="Path to configuration file"
	)

	parser.add_argument(
	"--model_name", type=str, default="gpt2", help="Model name to use"
	)

	parser.add_argument(
	"--num_problems", type=int, default=5, help="Number of problems to solve"
	)

	parser.add_argument(
	"--difficulty",
	type=str,
	default="medium",
	choices=["easy", "medium", "hard"],
	help="Difficulty level",
	)

	parser.add_argument(
	"--problem_type",
	type=str,
	default="arithmetic",
	choices=["arithmetic", "algebra", "geometry", "calculus"],
	help="Problem type",
	)

	parser.add_argument(
	"--device", type=str, default="auto", help="Device to use (cuda, cpu, auto)"
	)

	parser.add_argument("--interactive", action="store_true", help="Interactive mode")

	parser.add_argument(
	"--show_solutions", action="store_true", help="Show detailed solutions"
	)

	parser.add_argument(
	"--log_level",
	type=str,
	default="INFO",
	choices=["DEBUG", "INFO", "WARNING", "ERROR"],
	help="Logging level",
	)

	return parser.parse_args()

	def create_sample_problems():
	problems = [
	{
	"problem": "Solve: 2x + 5 = 13",
	"type": "algebra",
	"difficulty": "easy",
	"expected_answer": "x = 4",
	},
	{
	"problem": "What is the area of a circle with radius 5?",
	"type": "geometry",
	"difficulty": "medium",
	"expected_answer": "78.54 (approximately)",
	},
	{
	"problem": "Calculate: 15 × 23 + 7 × 8",
	"type": "arithmetic",
	"difficulty": "easy",
	"expected_answer": "409",
	},
	{
	"problem": "Find the derivative of x² + 3x + 2",
	"type": "calculus",
	"difficulty": "medium",
	"expected_answer": "2x + 3",
	},
	{
	"problem": "Solve the quadratic equation: x² - 5x + 6 = 0",
	"type": "algebra",
	"difficulty": "medium",
	"expected_answer": "x = 2 or x = 3",
	},
	]
	return problems

	def run_demo_problem(
	agent: MathReasoningAgent,
	environment: MathEnvironment,
	problem_info: dict,
	show_solution: bool = False,
	):
	problem = problem_info["problem"]
	problem_type = problem_info["type"]
	difficulty = problem_info["difficulty"]
	expected = problem_info.get("expected_answer", "Unknown")

	print(f"\n{'='*60}")
	print(f"Problem: {problem}")
	print(f"Type: {problem_type}, Difficulty: {difficulty}")
	print(f"Expected Answer: {expected}")
	print(f"{'='*60}")


	start_time = time.time()
	solution = agent.generate_solution(problem)
	generation_time = time.time() - start_time

	print(f"\nGenerated Solution:")
	print(f"{solution}")

	if show_solution:
	print(f"\nDetailed Analysis:")
	print(f"Generation time: {generation_time:.2f}s")
	print(f"Solution length: {len(solution)} characters")


	reward = environment.verify_solution(problem, solution)
	success = reward > 0

	print(f"\nVerification:")
	print(f"Reward: {reward:.2f}")
	print(f"Success: {'✅' if success else '❌'}")

	return {
	"problem": problem,
	"solution": solution,
	"reward": reward,
	"success": success,
	"generation_time": generation_time,
	"expected": expected,
	}

	def interactive_mode(agent: MathReasoningAgent, environment: MathEnvironment):
	print("\n" + "=" * 60)
	print("INTERACTIVE MODE")
	print("=" * 60)
	print("Enter math problems to solve. Type 'quit' to exit.")
	print("Available problem types: arithmetic, algebra, geometry, calculus")
	print("Available difficulties: easy, medium, hard")
	print("=" * 60)

	while True:
	try:
	print("\n" + "-" * 40)
	problem = input("Enter a math problem: ").strip()

	if problem.lower() in ["quit", "exit", "q"]:
	print("Goodbye!")
	break

	if not problem:
	continue


	print(f"\nSolving: {problem}")
	start_time = time.time()
	solution = agent.generate_solution(problem)
	generation_time = time.time() - start_time

	print(f"\nSolution:")
	print(f"{solution}")


	reward = environment.verify_solution(problem, solution)
	success = reward > 0

	print(f"\nVerification:")
	print(f"Reward: {reward:.2f}")
	print(f"Success: {'✅' if success else '❌'}")
	print(f"Time: {generation_time:.2f}s")

	except KeyboardInterrupt:
	print("\nGoodbye!")
	break
	except Exception as e:
	print(f"Error: {e}")

	def main():
	args = parse_args()


	log_file = "demo.log"
	setup_logging(log_file, args.log_level)

	logger.info("Starting RL-LLM demo")
	logger.info(f"Arguments: {args}")

	try:

	if os.path.exists(args.config):
	config = Config.from_yaml(args.config)
	logger.info(f"Loaded configuration from {args.config}")
	else:
	logger.warning(
	f"Configuration file {args.config} not found, using defaults"
	)
	config = Config()


	config.model.model_name = args.model_name
	config.model.device = args.device


	logger.info("Initializing agent and environment")
	agent = MathReasoningAgent(config.model)
	environment = MathEnvironment(config.environment)

	print("=" * 60)
	print("RL-LLM MATH REASONING DEMO")
	print("=" * 60)
	print(f"Model: {args.model_name}")
	print(f"Device: {config.model.device}")
	print(
	f"Agent initialized with {len(agent.few_shot_examples)} few-shot examples"
	)

	if args.interactive:
	interactive_mode(agent, environment)
	else:

	if args.num_problems > 0:
	print(f"\nRunning {args.num_problems} demo problems...")


	problems = []
	for i in range(args.num_problems):
	problem = environment.generate_problem(
	args.difficulty, args.problem_type
	)
	problems.append(
	{
	"problem": problem,
	"type": args.problem_type,
	"difficulty": args.difficulty,
	}
	)


	results = []
	for i, problem_info in enumerate(problems):
	print(f"\nProblem {i+1}/{args.num_problems}")
	result = run_demo_problem(
	agent, environment, problem_info, args.show_solutions
	)
	results.append(result)


	print(f"\n{'='*60}")
	print("DEMO SUMMARY")
	print(f"{'='*60}")
	print(f"Total problems: {len(results)}")
	print(f"Successful: {sum(1 for r in results if r['success'])}")
	print(
	f"Success rate: {sum(1 for r in results if r['success']) / len(results):.2%}"
	)
	print(
	f"Average reward: {sum(r['reward'] for r in results) / len(results):.2f}"
	)
	print(
	f"Average time: {sum(r['generation_time'] for r in results) / len(results):.2f}s"
	)


	print(f"\n{'='*60}")
	print("SAMPLE PROBLEMS")
	print(f"{'='*60}")
	sample_problems = create_sample_problems()

	for i, problem_info in enumerate(sample_problems[:3]):
	print(f"\nSample {i+1}:")
	print(f"Problem: {problem_info['problem']}")
	print(
	f"Type: {problem_info['type']}, Difficulty: {problem_info['difficulty']}"
	)
	print(f"Expected: {problem_info['expected_answer']}")

	logger.info("Demo completed successfully")

	except Exception as e:
	logger.error(f"Demo failed: {e}")
	print(f"\nError: {e}")
	raise

	if __name__ == "__main__":
	main()