Buckets:

tahamajs
/

Sysmem2_in_AI

Files

xet

tahamajs/Sysmem2_in_AI / ComputerAssignments /CA4_visual_question_answering /scripts /main.py

tahamajs

4 months ago

download

raw

23.2 kB

import torch
import torch.nn as nn
import argparse
import json
import os
import time
from typing import Dict, Any, List
import numpy as np
from pathlib import Path


from src.config.config import VQAConfig, load_config, create_default_config
from src.data.dataset import VQADataModule
from src.models.neural_module_networks import NeuralModuleNetwork
from src.models.scene_graph_models import SceneGraphVQAModel
from src.models.hybrid_vqa import HybridVQASystem
from src.evaluation.metrics import VQAEvaluator
from src.visualization.visualizer import (
    AttentionVisualizer,
    SceneGraphVisualizer,
    ResultsVisualizer,
)
from src.demo.demo import VQADemo, VQABenchmark
from src.utils.base_classes import set_seed, get_device_info


try:
    from src.llm.gemini_manager import (
        GeminiManager,
        GeminiConfig,
        GeminiVQAAgent,
        create_gemini_manager,
    )
    from src.llm.langgraph_workflow import VQAWorkflowGraph
    from src.llm.advanced_agents import (
        AdvancedVQAAgent,
        AgentConfig,
        MultiAgentVQASystem,
    )

    LLM_FEATURES_AVAILABLE = True
except ImportError as e:
    print(f"⚠️ Advanced LLM features not available: {e}")
    LLM_FEATURES_AVAILABLE = False

class VQATrainer:

    def __init__(self, config: VQAConfig):
        self.config = config
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")


        set_seed(config.seed)


        self.model = self._create_model()
        self.model.to(self.device)


        self.data_module = VQADataModule(config.data.__dict__)
        self.train_loader, self.val_loader, self.test_loader = (
            self.data_module.get_dataloaders()
        )


        self.evaluator = VQAEvaluator(config.evaluation.metrics)


        self.attention_viz = AttentionVisualizer()
        self.scene_viz = SceneGraphVisualizer()
        self.results_viz = ResultsVisualizer()

        print(f"✅ سیستم VQA آماده شد")
        print(f"مدل: {config.model.model_type}")
        print(f"دستگاه: {self.device}")
        print(f"نمونه‌های آموزشی: {len(self.train_loader.dataset)}")

    def _create_model(self) -> nn.Module:
        model_type = self.config.model.model_type

        if model_type == "nmn":
            return NeuralModuleNetwork(
                visual_dim=self.config.model.visual_dim,
                text_dim=self.config.model.text_dim,
                vocab_size=self.config.data.vocab_size,
            )
        elif model_type == "scene_graph":
            return SceneGraphVQAModel(
                visual_dim=self.config.model.visual_dim,
                text_dim=self.config.model.text_dim,
                vocab_size=self.config.data.vocab_size,
            )
        elif model_type == "hybrid_vqa":
            return HybridVQASystem(
                visual_dim=self.config.model.visual_dim,
                text_dim=self.config.model.text_dim,
                vocab_size=self.config.data.vocab_size,
            )
        else:
            raise ValueError(f"Unsupported model type: {model_type}")

    def train_epoch(self, epoch: int) -> Dict[str, float]:
        self.model.train()
        total_loss = 0.0
        num_batches = 0

        print(f"\n📚 آموزش Epoch {epoch}")

        for batch_idx, batch in enumerate(self.train_loader):

            batch_size = len(batch)
            visual_features = torch.randn(
                batch_size, self.config.model.visual_dim, 14, 14
            ).to(self.device)
            question_tokens = torch.randint(0, 1000, (batch_size, 20)).to(self.device)
            question_embedding = torch.randn(
                batch_size, 20, self.config.model.text_dim
            ).to(self.device)
            answer_labels = torch.randint(
                0, self.config.data.vocab_size, (batch_size,)
            ).to(self.device)


            if self.config.model.model_type == "nmn":
                result = self.model(visual_features, "test question")
                if "result" in result and isinstance(result["result"], torch.Tensor):
                    logits = result["result"]
                else:
                    logits = torch.randn(batch_size, self.config.data.vocab_size).to(
                        self.device
                    )
            else:
                result = self.model(
                    visual_features, question_tokens, question_embedding
                )
                logits = result["answer_logits"]


            loss = nn.CrossEntropyLoss()(logits, answer_labels)


            loss.backward()

            total_loss += loss.item()
            num_batches += 1

            if batch_idx % 10 == 0:
                print(
                    f"  Batch {batch_idx}/{len(self.train_loader)}: Loss = {loss.item():.4f}"
                )

        avg_loss = total_loss / num_batches
        print(f"✅ Epoch {epoch} تکمیل شد - میانگین Loss: {avg_loss:.4f}")

        return {"train_loss": avg_loss}

    def validate(self) -> Dict[str, float]:
        self.model.eval()

        predictions = []
        ground_truths = []
        confidences = []

        print(f"\n🔍 اعتبارسنجی")

        with torch.no_grad():
            for batch_idx, batch in enumerate(self.val_loader):
                batch_size = len(batch)
                visual_features = torch.randn(
                    batch_size, self.config.model.visual_dim, 14, 14
                ).to(self.device)
                question_tokens = torch.randint(0, 1000, (batch_size, 20)).to(
                    self.device
                )
                question_embedding = torch.randn(
                    batch_size, 20, self.config.model.text_dim
                ).to(self.device)


                if self.config.model.model_type == "nmn":
                    result = self.model(visual_features, "test question")
                    if "result" in result and isinstance(
                        result["result"], torch.Tensor
                    ):
                        logits = result["result"]
                    else:
                        logits = torch.randn(
                            batch_size, self.config.data.vocab_size
                        ).to(self.device)
                else:
                    result = self.model(
                        visual_features, question_tokens, question_embedding
                    )
                    logits = result["answer_logits"]


                predicted_ids = torch.argmax(logits, dim=1)
                predicted_answers = [f"answer_{id.item()}" for id in predicted_ids]
                batch_confidences = (
                    torch.max(torch.softmax(logits, dim=1), dim=1)[0].cpu().numpy()
                )


                ground_truth_answers = [sample.answer for sample in batch]

                predictions.extend(predicted_answers)
                ground_truths.extend(ground_truth_answers)
                confidences.extend(batch_confidences)

                if batch_idx % 5 == 0:
                    print(f"  Batch {batch_idx}/{len(self.val_loader)}")


        eval_results = self.evaluator.evaluate(predictions, ground_truths, confidences)

        print(f"✅ اعتبارسنجی تکمیل شد")
        for metric, score in eval_results.items():
            print(f"  {metric}: {score:.4f}")

        return eval_results

    def train(self):
        print(f"\n🚀 شروع آموزش")
        print(f"تعداد Epoch ها: {self.config.training.num_epochs}")

        train_losses = []
        val_metrics = []

        for epoch in range(1, self.config.training.num_epochs + 1):

            train_results = self.train_epoch(epoch)
            train_losses.append(train_results["train_loss"])


            if epoch % 5 == 0:
                val_results = self.validate()
                val_metrics.append(val_results)


            if epoch % self.config.training.checkpoint_frequency == 0:
                self.save_checkpoint(epoch)


        self.visualize_training_results(train_losses, val_metrics)

        print(f"\n✅ آموزش تکمیل شد!")

    def save_checkpoint(self, epoch: int):
        checkpoint_dir = Path(self.config.checkpoint_dir)
        checkpoint_dir.mkdir(exist_ok=True)

        checkpoint_path = checkpoint_dir / f"checkpoint_epoch_{epoch}.pt"

        checkpoint = {
            "epoch": epoch,
            "model_state_dict": self.model.state_dict(),
            "config": self.config.__dict__,
            "timestamp": time.time(),
        }

        torch.save(checkpoint, checkpoint_path)
        print(f"💾 Checkpoint در {checkpoint_path} ذخیره شد")

    def visualize_training_results(
        self, train_losses: List[float], val_metrics: List[Dict[str, float]]
    ):
        viz_dir = Path("visualizations")
        viz_dir.mkdir(exist_ok=True)


        self.results_viz.plot_training_curves(
            train_losses, save_path=str(viz_dir / "training_curves.png")
        )

        print(f"📊 نتایج آموزش در {viz_dir} ذخیره شدند")

    def test(self):
        print(f"\n🧪 تست مدل")

        self.model.eval()

        predictions = []
        ground_truths = []
        confidences = []

        with torch.no_grad():
            for batch_idx, batch in enumerate(self.test_loader):
                batch_size = len(batch)
                visual_features = torch.randn(
                    batch_size, self.config.model.visual_dim, 14, 14
                ).to(self.device)
                question_tokens = torch.randint(0, 1000, (batch_size, 20)).to(
                    self.device
                )
                question_embedding = torch.randn(
                    batch_size, 20, self.config.model.text_dim
                ).to(self.device)


                if self.config.model.model_type == "nmn":
                    result = self.model(visual_features, "test question")
                    if "result" in result and isinstance(
                        result["result"], torch.Tensor
                    ):
                        logits = result["result"]
                    else:
                        logits = torch.randn(
                            batch_size, self.config.data.vocab_size
                        ).to(self.device)
                else:
                    result = self.model(
                        visual_features, question_tokens, question_embedding
                    )
                    logits = result["answer_logits"]


                predicted_ids = torch.argmax(logits, dim=1)
                predicted_answers = [f"answer_{id.item()}" for id in predicted_ids]
                batch_confidences = (
                    torch.max(torch.softmax(logits, dim=1), dim=1)[0].cpu().numpy()
                )


                ground_truth_answers = [sample.answer for sample in batch]

                predictions.extend(predicted_answers)
                ground_truths.extend(ground_truth_answers)
                confidences.extend(batch_confidences)


        final_results = self.evaluator.evaluate(predictions, ground_truths, confidences)

        print(f"✅ تست تکمیل شد")
        print(f"نتایج نهایی:")
        for metric, score in final_results.items():
            print(f"  {metric}: {score:.4f}")

        return final_results

class AdvancedVQASystem:

    def __init__(self, config: VQAConfig):
        self.config = config
        self.gemini_manager = None
        self.workflow_graph = None
        self.multi_agent_system = None

        if LLM_FEATURES_AVAILABLE:
            self._setup_llm_features()

    def _setup_llm_features(self):
        try:

            api_key = os.getenv("GOOGLE_API_KEY")
            if api_key:
                self.gemini_manager = create_gemini_manager(api_key)
                print("✅ Gemini Manager initialized")
            else:
                print("⚠️ GOOGLE_API_KEY not set, Gemini features disabled")


            self.workflow_graph = VQAWorkflowGraph(self.gemini_manager)
            print("✅ LangGraph Workflow initialized")


            self.multi_agent_system = MultiAgentVQASystem(self.gemini_manager)
            print("✅ Multi-Agent System initialized")

        except Exception as e:
            print(f"⚠️ Error setting up LLM features: {e}")

    def process_with_gemini(self, question: str, image_path: str) -> Dict[str, Any]:
        if not self.gemini_manager:
            return {
                "answer": "Gemini Manager در دسترس نیست",
                "success": False,
                "error": "Gemini not available",
            }

        try:
            result = self.gemini_manager.ask_question_with_image(question, image_path)
            return {
                "answer": result["answer"],
                "reasoning": result["reasoning"],
                "confidence": result["confidence"],
                "method": "gemini",
                "success": True,
            }
        except Exception as e:
            return {
                "answer": f"خطا در Gemini: {str(e)}",
                "success": False,
                "error": str(e),
            }

    def process_with_workflow(self, question: str, image_path: str) -> Dict[str, Any]:
        if not self.workflow_graph:
            return {
                "answer": "Workflow Graph در دسترس نیست",
                "success": False,
                "error": "Workflow not available",
            }

        try:
            result = self.workflow_graph.process_vqa(question, image_path)
            return {
                "answer": result["answer"],
                "reasoning_steps": result["reasoning_steps"],
                "confidence": result["confidence"],
                "method": "langgraph_workflow",
                "success": result["success"],
            }
        except Exception as e:
            return {
                "answer": f"خطا در Workflow: {str(e)}",
                "success": False,
                "error": str(e),
            }

    def process_with_agents(
        self, question: str, image_path: str, agent_type: str = "comprehensive"
    ) -> Dict[str, Any]:
        if not self.multi_agent_system:
            return {
                "answer": "Multi-Agent System در دسترس نیست",
                "success": False,
                "error": "Agents not available",
            }

        try:
            if agent_type == "all":
                result = self.multi_agent_system.process_with_all_agents(
                    question, image_path
                )
            else:
                result = self.multi_agent_system.process_with_specific_agent(
                    agent_type, question, image_path
                )

            return {
                "answer": result["answer"],
                "method": f"multi_agent_{agent_type}",
                "success": result["success"],
                "agent_info": result.get("best_agent", agent_type),
            }
        except Exception as e:
            return {
                "answer": f"خطا در Agent ها: {str(e)}",
                "success": False,
                "error": str(e),
            }

    def process_comprehensive(self, question: str, image_path: str) -> Dict[str, Any]:
        results = {}


        if self.gemini_manager:
            results["gemini"] = self.process_with_gemini(question, image_path)


        if self.workflow_graph:
            results["workflow"] = self.process_with_workflow(question, image_path)


        if self.multi_agent_system:
            results["agents"] = self.process_with_agents(
                question, image_path, "comprehensive"
            )


        successful_results = [r for r in results.values() if r.get("success", False)]

        if successful_results:

            best_result = max(
                successful_results, key=lambda x: x.get("confidence", 0.5)
            )
            return {
                "answer": best_result["answer"],
                "method": best_result["method"],
                "confidence": best_result.get("confidence", 0.5),
                "all_results": results,
                "success": True,
            }
        else:
            return {
                "answer": "هیچ روشی موفق به پردازش نشد",
                "all_results": results,
                "success": False,
            }

    def get_system_info(self) -> Dict[str, Any]:
        info = {
            "llm_features_available": LLM_FEATURES_AVAILABLE,
            "gemini_available": self.gemini_manager is not None,
            "workflow_available": self.workflow_graph is not None,
            "agents_available": self.multi_agent_system is not None,
        }

        if self.gemini_manager:
            info["gemini_info"] = self.gemini_manager.get_model_info()

        if self.workflow_graph:
            info["workflow_info"] = self.workflow_graph.get_graph_info()

        if self.multi_agent_system:
            info["agents_info"] = self.multi_agent_system.get_system_info()

        return info

def main():
    parser = argparse.ArgumentParser(description="سیستم VQA نوروسیمبولیک")
    parser.add_argument(
        "--mode",
        type=str,
        default="demo",
        choices=[
            "train",
            "test",
            "demo",
            "benchmark",
            "gemini",
            "workflow",
            "agents",
            "comprehensive",
        ],
        help="حالت اجرا",
    )
    parser.add_argument(
        "--config",
        type=str,
        default="src/config/default_config.yaml",
        help="مسیر فایل تنظیمات",
    )
    parser.add_argument(
        "--model",
        type=str,
        default="hybrid_vqa",
        choices=["nmn", "scene_graph", "hybrid_vqa"],
        help="نوع مدل",
    )
    parser.add_argument("--epochs", type=int, default=10, help="تعداد epoch ها")
    parser.add_argument(
        "--samples", type=int, default=50, help="تعداد نمونه‌ها برای بنچمارک"
    )
    parser.add_argument("--image", type=str, help="مسیر تصویر برای پردازش")
    parser.add_argument("--question", type=str, help="سوال برای پردازش")
    parser.add_argument(
        "--agent-type",
        type=str,
        default="comprehensive",
        choices=["image", "question", "reasoning", "answer", "comprehensive", "all"],
        help="نوع Agent برای استفاده",
    )

    args = parser.parse_args()

    print("🎯 سیستم VQA نوروسیمبولیک")
    print("=" * 50)


    device_info = get_device_info()
    print(f"اطلاعات سیستم:")
    print(f"  PyTorch: {device_info['pytorch_version']}")
    print(f"  CUDA: {'✅' if device_info['cuda_available'] else '❌'}")
    if device_info["cuda_available"]:
        print(f"  دستگاه CUDA: {device_info['cuda_device_name']}")


    if os.path.exists(args.config):
        config = load_config(args.config)
    else:
        print(f"⚠️ فایل تنظیمات {args.config} یافت نشد. ایجاد تنظیمات پیش‌فرض...")
        create_default_config(args.config)
        config = load_config(args.config)


    config.model.model_type = args.model
    config.training.num_epochs = args.epochs


    if args.mode == "train":
        print(f"\n🚀 شروع آموزش مدل {args.model}")
        trainer = VQATrainer(config)
        trainer.train()

    elif args.mode == "test":
        print(f"\n🧪 تست مدل {args.model}")
        trainer = VQATrainer(config)
        results = trainer.test()

    elif args.mode == "demo":
        print(f"\n🎮 اجرای دمو")
        demo = VQADemo(args.model)
        demo.interactive_demo()

    elif args.mode == "benchmark":
        print(f"\n🏁 اجرای بنچمارک")
        benchmark = VQABenchmark()
        results = benchmark.compare_all_models(num_samples=args.samples)
        benchmark.generate_report(results)

    elif args.mode == "gemini":
        print(f"\n🤖 اجرای Gemini VQA")
        if not args.image or not args.question:
            print("❌ لطفاً مسیر تصویر و سوال را مشخص کنید")
            return

        advanced_system = AdvancedVQASystem(config)
        result = advanced_system.process_with_gemini(args.question, args.image)

        print(f"سوال: {args.question}")
        print(f"پاسخ: {result['answer']}")
        print(f"اعتماد: {result.get('confidence', 0):.2f}")
        if result.get("reasoning"):
            print(f"استدلال: {result['reasoning']}")

    elif args.mode == "workflow":
        print(f"\n🔄 اجرای LangGraph Workflow")
        if not args.image or not args.question:
            print("❌ لطفاً مسیر تصویر و سوال را مشخص کنید")
            return

        advanced_system = AdvancedVQASystem(config)
        result = advanced_system.process_with_workflow(args.question, args.image)

        print(f"سوال: {args.question}")
        print(f"پاسخ: {result['answer']}")
        print(f"اعتماد: {result.get('confidence', 0):.2f}")
        if result.get("reasoning_steps"):
            print(f"مراحل استدلال:")
            for i, step in enumerate(result["reasoning_steps"], 1):
                print(f"  {i}. {step}")

    elif args.mode == "agents":
        print(f"\n👥 اجرای Multi-Agent System")
        if not args.image or not args.question:
            print("❌ لطفاً مسیر تصویر و سوال را مشخص کنید")
            return

        advanced_system = AdvancedVQASystem(config)
        result = advanced_system.process_with_agents(
            args.question, args.image, args.agent_type
        )

        print(f"سوال: {args.question}")
        print(f"پاسخ: {result['answer']}")
        print(f"Agent استفاده شده: {result.get('agent_info', 'unknown')}")

    elif args.mode == "comprehensive":
        print(f"\n🌟 اجرای پردازش جامع")
        if not args.image or not args.question:
            print("❌ لطفاً مسیر تصویر و سوال را مشخص کنید")
            return

        advanced_system = AdvancedVQASystem(config)
        result = advanced_system.process_comprehensive(args.question, args.image)

        print(f"سوال: {args.question}")
        print(f"پاسخ نهایی: {result['answer']}")
        print(f"روش استفاده شده: {result['method']}")
        print(f"اعتماد: {result.get('confidence', 0):.2f}")

        if result.get("all_results"):
            print(f"\nنتایج تمام روش‌ها:")
            for method, method_result in result["all_results"].items():
                print(f"  {method}: {method_result.get('success', False)}")

    else:
        print(f"❌ حالت نامعتبر: {args.mode}")
        return

    print(f"\n✅ برنامه با موفقیت تکمیل شد!")

if __name__ == "__main__":
    main()

Xet Storage Details

Size:: 23.2 kB
Xet hash:: de9322b45e2eaa00968eef0cc83db9d9476d72dd372c18c333b029748c5c3411

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

	import torch
	import torch.nn as nn
	import argparse
	import json
	import os
	import time
	from typing import Dict, Any, List
	import numpy as np
	from pathlib import Path


	from src.config.config import VQAConfig, load_config, create_default_config
	from src.data.dataset import VQADataModule
	from src.models.neural_module_networks import NeuralModuleNetwork
	from src.models.scene_graph_models import SceneGraphVQAModel
	from src.models.hybrid_vqa import HybridVQASystem
	from src.evaluation.metrics import VQAEvaluator
	from src.visualization.visualizer import (
	AttentionVisualizer,
	SceneGraphVisualizer,
	ResultsVisualizer,
	)
	from src.demo.demo import VQADemo, VQABenchmark
	from src.utils.base_classes import set_seed, get_device_info


	try:
	from src.llm.gemini_manager import (
	GeminiManager,
	GeminiConfig,
	GeminiVQAAgent,
	create_gemini_manager,
	)
	from src.llm.langgraph_workflow import VQAWorkflowGraph
	from src.llm.advanced_agents import (
	AdvancedVQAAgent,
	AgentConfig,
	MultiAgentVQASystem,
	)

	LLM_FEATURES_AVAILABLE = True
	except ImportError as e:
	print(f"⚠️ Advanced LLM features not available: {e}")
	LLM_FEATURES_AVAILABLE = False

	class VQATrainer:

	def __init__(self, config: VQAConfig):
	self.config = config
	self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")


	set_seed(config.seed)


	self.model = self._create_model()
	self.model.to(self.device)


	self.data_module = VQADataModule(config.data.__dict__)
	self.train_loader, self.val_loader, self.test_loader = (
	self.data_module.get_dataloaders()
	)


	self.evaluator = VQAEvaluator(config.evaluation.metrics)


	self.attention_viz = AttentionVisualizer()
	self.scene_viz = SceneGraphVisualizer()
	self.results_viz = ResultsVisualizer()

	print(f"✅ سیستم VQA آماده شد")
	print(f"مدل: {config.model.model_type}")
	print(f"دستگاه: {self.device}")
	print(f"نمونه‌های آموزشی: {len(self.train_loader.dataset)}")

	def _create_model(self) -> nn.Module:
	model_type = self.config.model.model_type

	if model_type == "nmn":
	return NeuralModuleNetwork(
	visual_dim=self.config.model.visual_dim,
	text_dim=self.config.model.text_dim,
	vocab_size=self.config.data.vocab_size,
	)
	elif model_type == "scene_graph":
	return SceneGraphVQAModel(
	visual_dim=self.config.model.visual_dim,
	text_dim=self.config.model.text_dim,
	vocab_size=self.config.data.vocab_size,
	)
	elif model_type == "hybrid_vqa":
	return HybridVQASystem(
	visual_dim=self.config.model.visual_dim,
	text_dim=self.config.model.text_dim,
	vocab_size=self.config.data.vocab_size,
	)
	else:
	raise ValueError(f"Unsupported model type: {model_type}")

	def train_epoch(self, epoch: int) -> Dict[str, float]:
	self.model.train()
	total_loss = 0.0
	num_batches = 0

	print(f"\n📚 آموزش Epoch {epoch}")

	for batch_idx, batch in enumerate(self.train_loader):

	batch_size = len(batch)
	visual_features = torch.randn(
	batch_size, self.config.model.visual_dim, 14, 14
	).to(self.device)
	question_tokens = torch.randint(0, 1000, (batch_size, 20)).to(self.device)
	question_embedding = torch.randn(
	batch_size, 20, self.config.model.text_dim
	).to(self.device)
	answer_labels = torch.randint(
	0, self.config.data.vocab_size, (batch_size,)
	).to(self.device)


	if self.config.model.model_type == "nmn":
	result = self.model(visual_features, "test question")
	if "result" in result and isinstance(result["result"], torch.Tensor):
	logits = result["result"]
	else:
	logits = torch.randn(batch_size, self.config.data.vocab_size).to(
	self.device
	)
	else:
	result = self.model(
	visual_features, question_tokens, question_embedding
	)
	logits = result["answer_logits"]


	loss = nn.CrossEntropyLoss()(logits, answer_labels)


	loss.backward()

	total_loss += loss.item()
	num_batches += 1

	if batch_idx % 10 == 0:
	print(
	f" Batch {batch_idx}/{len(self.train_loader)}: Loss = {loss.item():.4f}"
	)

	avg_loss = total_loss / num_batches
	print(f"✅ Epoch {epoch} تکمیل شد - میانگین Loss: {avg_loss:.4f}")

	return {"train_loss": avg_loss}

	def validate(self) -> Dict[str, float]:
	self.model.eval()

	predictions = []
	ground_truths = []
	confidences = []

	print(f"\n🔍 اعتبارسنجی")

	with torch.no_grad():
	for batch_idx, batch in enumerate(self.val_loader):
	batch_size = len(batch)
	visual_features = torch.randn(
	batch_size, self.config.model.visual_dim, 14, 14
	).to(self.device)
	question_tokens = torch.randint(0, 1000, (batch_size, 20)).to(
	self.device
	)
	question_embedding = torch.randn(
	batch_size, 20, self.config.model.text_dim
	).to(self.device)


	if self.config.model.model_type == "nmn":
	result = self.model(visual_features, "test question")
	if "result" in result and isinstance(
	result["result"], torch.Tensor
	):
	logits = result["result"]
	else:
	logits = torch.randn(
	batch_size, self.config.data.vocab_size
	).to(self.device)
	else:
	result = self.model(
	visual_features, question_tokens, question_embedding
	)
	logits = result["answer_logits"]


	predicted_ids = torch.argmax(logits, dim=1)
	predicted_answers = [f"answer_{id.item()}" for id in predicted_ids]
	batch_confidences = (
	torch.max(torch.softmax(logits, dim=1), dim=1)[0].cpu().numpy()
	)


	ground_truth_answers = [sample.answer for sample in batch]

	predictions.extend(predicted_answers)
	ground_truths.extend(ground_truth_answers)
	confidences.extend(batch_confidences)

	if batch_idx % 5 == 0:
	print(f" Batch {batch_idx}/{len(self.val_loader)}")


	eval_results = self.evaluator.evaluate(predictions, ground_truths, confidences)

	print(f"✅ اعتبارسنجی تکمیل شد")
	for metric, score in eval_results.items():
	print(f" {metric}: {score:.4f}")

	return eval_results

	def train(self):
	print(f"\n🚀 شروع آموزش")
	print(f"تعداد Epoch ها: {self.config.training.num_epochs}")

	train_losses = []
	val_metrics = []

	for epoch in range(1, self.config.training.num_epochs + 1):

	train_results = self.train_epoch(epoch)
	train_losses.append(train_results["train_loss"])


	if epoch % 5 == 0:
	val_results = self.validate()
	val_metrics.append(val_results)


	if epoch % self.config.training.checkpoint_frequency == 0:
	self.save_checkpoint(epoch)


	self.visualize_training_results(train_losses, val_metrics)

	print(f"\n✅ آموزش تکمیل شد!")

	def save_checkpoint(self, epoch: int):
	checkpoint_dir = Path(self.config.checkpoint_dir)
	checkpoint_dir.mkdir(exist_ok=True)

	checkpoint_path = checkpoint_dir / f"checkpoint_epoch_{epoch}.pt"

	checkpoint = {
	"epoch": epoch,
	"model_state_dict": self.model.state_dict(),
	"config": self.config.__dict__,
	"timestamp": time.time(),
	}

	torch.save(checkpoint, checkpoint_path)
	print(f"💾 Checkpoint در {checkpoint_path} ذخیره شد")

	def visualize_training_results(
	self, train_losses: List[float], val_metrics: List[Dict[str, float]]
	):
	viz_dir = Path("visualizations")
	viz_dir.mkdir(exist_ok=True)


	self.results_viz.plot_training_curves(
	train_losses, save_path=str(viz_dir / "training_curves.png")
	)

	print(f"📊 نتایج آموزش در {viz_dir} ذخیره شدند")

	def test(self):
	print(f"\n🧪 تست مدل")

	self.model.eval()

	predictions = []
	ground_truths = []
	confidences = []

	with torch.no_grad():
	for batch_idx, batch in enumerate(self.test_loader):
	batch_size = len(batch)
	visual_features = torch.randn(
	batch_size, self.config.model.visual_dim, 14, 14
	).to(self.device)
	question_tokens = torch.randint(0, 1000, (batch_size, 20)).to(
	self.device
	)
	question_embedding = torch.randn(
	batch_size, 20, self.config.model.text_dim
	).to(self.device)


	if self.config.model.model_type == "nmn":
	result = self.model(visual_features, "test question")
	if "result" in result and isinstance(
	result["result"], torch.Tensor
	):
	logits = result["result"]
	else:
	logits = torch.randn(
	batch_size, self.config.data.vocab_size
	).to(self.device)
	else:
	result = self.model(
	visual_features, question_tokens, question_embedding
	)
	logits = result["answer_logits"]


	predicted_ids = torch.argmax(logits, dim=1)
	predicted_answers = [f"answer_{id.item()}" for id in predicted_ids]
	batch_confidences = (
	torch.max(torch.softmax(logits, dim=1), dim=1)[0].cpu().numpy()
	)


	ground_truth_answers = [sample.answer for sample in batch]

	predictions.extend(predicted_answers)
	ground_truths.extend(ground_truth_answers)
	confidences.extend(batch_confidences)


	final_results = self.evaluator.evaluate(predictions, ground_truths, confidences)

	print(f"✅ تست تکمیل شد")
	print(f"نتایج نهایی:")
	for metric, score in final_results.items():
	print(f" {metric}: {score:.4f}")

	return final_results

	class AdvancedVQASystem:

	def __init__(self, config: VQAConfig):
	self.config = config
	self.gemini_manager = None
	self.workflow_graph = None
	self.multi_agent_system = None

	if LLM_FEATURES_AVAILABLE:
	self._setup_llm_features()

	def _setup_llm_features(self):
	try:

	api_key = os.getenv("GOOGLE_API_KEY")
	if api_key:
	self.gemini_manager = create_gemini_manager(api_key)
	print("✅ Gemini Manager initialized")
	else:
	print("⚠️ GOOGLE_API_KEY not set, Gemini features disabled")


	self.workflow_graph = VQAWorkflowGraph(self.gemini_manager)
	print("✅ LangGraph Workflow initialized")


	self.multi_agent_system = MultiAgentVQASystem(self.gemini_manager)
	print("✅ Multi-Agent System initialized")

	except Exception as e:
	print(f"⚠️ Error setting up LLM features: {e}")

	def process_with_gemini(self, question: str, image_path: str) -> Dict[str, Any]:
	if not self.gemini_manager:
	return {
	"answer": "Gemini Manager در دسترس نیست",
	"success": False,
	"error": "Gemini not available",
	}

	try:
	result = self.gemini_manager.ask_question_with_image(question, image_path)
	return {
	"answer": result["answer"],
	"reasoning": result["reasoning"],
	"confidence": result["confidence"],
	"method": "gemini",
	"success": True,
	}
	except Exception as e:
	return {
	"answer": f"خطا در Gemini: {str(e)}",
	"success": False,
	"error": str(e),
	}

	def process_with_workflow(self, question: str, image_path: str) -> Dict[str, Any]:
	if not self.workflow_graph:
	return {
	"answer": "Workflow Graph در دسترس نیست",
	"success": False,
	"error": "Workflow not available",
	}

	try:
	result = self.workflow_graph.process_vqa(question, image_path)
	return {
	"answer": result["answer"],
	"reasoning_steps": result["reasoning_steps"],
	"confidence": result["confidence"],
	"method": "langgraph_workflow",
	"success": result["success"],
	}
	except Exception as e:
	return {
	"answer": f"خطا در Workflow: {str(e)}",
	"success": False,
	"error": str(e),
	}

	def process_with_agents(
	self, question: str, image_path: str, agent_type: str = "comprehensive"
	) -> Dict[str, Any]:
	if not self.multi_agent_system:
	return {
	"answer": "Multi-Agent System در دسترس نیست",
	"success": False,
	"error": "Agents not available",
	}

	try:
	if agent_type == "all":
	result = self.multi_agent_system.process_with_all_agents(
	question, image_path
	)
	else:
	result = self.multi_agent_system.process_with_specific_agent(
	agent_type, question, image_path
	)

	return {
	"answer": result["answer"],
	"method": f"multi_agent_{agent_type}",
	"success": result["success"],
	"agent_info": result.get("best_agent", agent_type),
	}
	except Exception as e:
	return {
	"answer": f"خطا در Agent ها: {str(e)}",
	"success": False,
	"error": str(e),
	}

	def process_comprehensive(self, question: str, image_path: str) -> Dict[str, Any]:
	results = {}


	if self.gemini_manager:
	results["gemini"] = self.process_with_gemini(question, image_path)


	if self.workflow_graph:
	results["workflow"] = self.process_with_workflow(question, image_path)


	if self.multi_agent_system:
	results["agents"] = self.process_with_agents(
	question, image_path, "comprehensive"
	)


	successful_results = [r for r in results.values() if r.get("success", False)]

	if successful_results:

	best_result = max(
	successful_results, key=lambda x: x.get("confidence", 0.5)
	)
	return {
	"answer": best_result["answer"],
	"method": best_result["method"],
	"confidence": best_result.get("confidence", 0.5),
	"all_results": results,
	"success": True,
	}
	else:
	return {
	"answer": "هیچ روشی موفق به پردازش نشد",
	"all_results": results,
	"success": False,
	}

	def get_system_info(self) -> Dict[str, Any]:
	info = {
	"llm_features_available": LLM_FEATURES_AVAILABLE,
	"gemini_available": self.gemini_manager is not None,
	"workflow_available": self.workflow_graph is not None,
	"agents_available": self.multi_agent_system is not None,
	}

	if self.gemini_manager:
	info["gemini_info"] = self.gemini_manager.get_model_info()

	if self.workflow_graph:
	info["workflow_info"] = self.workflow_graph.get_graph_info()

	if self.multi_agent_system:
	info["agents_info"] = self.multi_agent_system.get_system_info()

	return info

	def main():
	parser = argparse.ArgumentParser(description="سیستم VQA نوروسیمبولیک")
	parser.add_argument(
	"--mode",
	type=str,
	default="demo",
	choices=[
	"train",
	"test",
	"demo",
	"benchmark",
	"gemini",
	"workflow",
	"agents",
	"comprehensive",
	],
	help="حالت اجرا",
	)
	parser.add_argument(
	"--config",
	type=str,
	default="src/config/default_config.yaml",
	help="مسیر فایل تنظیمات",
	)
	parser.add_argument(
	"--model",
	type=str,
	default="hybrid_vqa",
	choices=["nmn", "scene_graph", "hybrid_vqa"],
	help="نوع مدل",
	)
	parser.add_argument("--epochs", type=int, default=10, help="تعداد epoch ها")
	parser.add_argument(
	"--samples", type=int, default=50, help="تعداد نمونه‌ها برای بنچمارک"
	)
	parser.add_argument("--image", type=str, help="مسیر تصویر برای پردازش")
	parser.add_argument("--question", type=str, help="سوال برای پردازش")
	parser.add_argument(
	"--agent-type",
	type=str,
	default="comprehensive",
	choices=["image", "question", "reasoning", "answer", "comprehensive", "all"],
	help="نوع Agent برای استفاده",
	)

	args = parser.parse_args()

	print("🎯 سیستم VQA نوروسیمبولیک")
	print("=" * 50)


	device_info = get_device_info()
	print(f"اطلاعات سیستم:")
	print(f" PyTorch: {device_info['pytorch_version']}")
	print(f" CUDA: {'✅' if device_info['cuda_available'] else '❌'}")
	if device_info["cuda_available"]:
	print(f" دستگاه CUDA: {device_info['cuda_device_name']}")


	if os.path.exists(args.config):
	config = load_config(args.config)
	else:
	print(f"⚠️ فایل تنظیمات {args.config} یافت نشد. ایجاد تنظیمات پیش‌فرض...")
	create_default_config(args.config)
	config = load_config(args.config)


	config.model.model_type = args.model
	config.training.num_epochs = args.epochs


	if args.mode == "train":
	print(f"\n🚀 شروع آموزش مدل {args.model}")
	trainer = VQATrainer(config)
	trainer.train()

	elif args.mode == "test":
	print(f"\n🧪 تست مدل {args.model}")
	trainer = VQATrainer(config)
	results = trainer.test()

	elif args.mode == "demo":
	print(f"\n🎮 اجرای دمو")
	demo = VQADemo(args.model)
	demo.interactive_demo()

	elif args.mode == "benchmark":
	print(f"\n🏁 اجرای بنچمارک")
	benchmark = VQABenchmark()
	results = benchmark.compare_all_models(num_samples=args.samples)
	benchmark.generate_report(results)

	elif args.mode == "gemini":
	print(f"\n🤖 اجرای Gemini VQA")
	if not args.image or not args.question:
	print("❌ لطفاً مسیر تصویر و سوال را مشخص کنید")
	return

	advanced_system = AdvancedVQASystem(config)
	result = advanced_system.process_with_gemini(args.question, args.image)

	print(f"سوال: {args.question}")
	print(f"پاسخ: {result['answer']}")
	print(f"اعتماد: {result.get('confidence', 0):.2f}")
	if result.get("reasoning"):
	print(f"استدلال: {result['reasoning']}")

	elif args.mode == "workflow":
	print(f"\n🔄 اجرای LangGraph Workflow")
	if not args.image or not args.question:
	print("❌ لطفاً مسیر تصویر و سوال را مشخص کنید")
	return

	advanced_system = AdvancedVQASystem(config)
	result = advanced_system.process_with_workflow(args.question, args.image)

	print(f"سوال: {args.question}")
	print(f"پاسخ: {result['answer']}")
	print(f"اعتماد: {result.get('confidence', 0):.2f}")
	if result.get("reasoning_steps"):
	print(f"مراحل استدلال:")
	for i, step in enumerate(result["reasoning_steps"], 1):
	print(f" {i}. {step}")

	elif args.mode == "agents":
	print(f"\n👥 اجرای Multi-Agent System")
	if not args.image or not args.question:
	print("❌ لطفاً مسیر تصویر و سوال را مشخص کنید")
	return

	advanced_system = AdvancedVQASystem(config)
	result = advanced_system.process_with_agents(
	args.question, args.image, args.agent_type
	)

	print(f"سوال: {args.question}")
	print(f"پاسخ: {result['answer']}")
	print(f"Agent استفاده شده: {result.get('agent_info', 'unknown')}")

	elif args.mode == "comprehensive":
	print(f"\n🌟 اجرای پردازش جامع")
	if not args.image or not args.question:
	print("❌ لطفاً مسیر تصویر و سوال را مشخص کنید")
	return

	advanced_system = AdvancedVQASystem(config)
	result = advanced_system.process_comprehensive(args.question, args.image)

	print(f"سوال: {args.question}")
	print(f"پاسخ نهایی: {result['answer']}")
	print(f"روش استفاده شده: {result['method']}")
	print(f"اعتماد: {result.get('confidence', 0):.2f}")

	if result.get("all_results"):
	print(f"\nنتایج تمام روش‌ها:")
	for method, method_result in result["all_results"].items():
	print(f" {method}: {method_result.get('success', False)}")

	else:
	print(f"❌ حالت نامعتبر: {args.mode}")
	return

	print(f"\n✅ برنامه با موفقیت تکمیل شد!")

	if __name__ == "__main__":
	main()