Buckets:

tahamajs
/

Sysmem2_in_AI

Files

xet

tahamajs/Sysmem2_in_AI / ComputerAssignments /CA15_test_time_training /src /advanced_agents.py

tahamajs

5 months ago

download

raw

21.5 kB

import os
import json
import asyncio
from typing import List, Dict, Any, Optional, Union, Callable
from dataclasses import dataclass, field
from enum import Enum
from abc import ABC, abstractmethod
import time

try:
    from langchain.agents import AgentExecutor, create_react_agent
    from langchain.tools import BaseTool, Tool
    from langchain.schema import BaseMessage, HumanMessage, AIMessage, SystemMessage
    from langchain.llms import OpenAI
    from langchain.chat_models import ChatOpenAI
    LANCHAIN_AVAILABLE = True
except ImportError:
    LANCHAIN_AVAILABLE = False

try:
    import google.generativeai as genai
    GEMINI_AVAILABLE = True
except ImportError:
    GEMINI_AVAILABLE = False

class AgentType(Enum):
    PATTERN_ANALYST = "pattern_analyst"
    TRANSFORMATION_EXPERT = "transformation_expert"
    VERIFICATION_SPECIALIST = "verification_specialist"
    ERROR_HANDLER = "error_handler"
    COORDINATOR = "coordinator"
    CREATIVE_GENERATOR = "creative_generator"

class AgentStatus(Enum):
    IDLE = "idle"
    WORKING = "working"
    WAITING = "waiting"
    ERROR = "error"
    COMPLETED = "completed"

@dataclass
class AgentMessage:
    sender: str
    recipient: str
    message_type: str
    content: Dict[str, Any]
    timestamp: float = field(default_factory=time.time)
    priority: int = 1

@dataclass
class TaskContext:
    task_id: str
    task_description: str
    examples: List[Dict[str, str]]
    query: str
    current_step: str
    results: Dict[str, Any] = field(default_factory=dict)
    metadata: Dict[str, Any] = field(default_factory=dict)

class BaseTTTAgent(ABC):
    
    def __init__(self, agent_id: str, agent_type: AgentType):
        self.agent_id = agent_id
        self.agent_type = agent_type
        self.status = AgentStatus.IDLE
        self.message_queue = []
        self.capabilities = []
        self.specializations = []
        self._initialize_agent()
    
    @abstractmethod
    def _initialize_agent(self):
        pass
    
    @abstractmethod
    async def process_task(self, context: TaskContext) -> Dict[str, Any]:
        pass
    
    def send_message(self, recipient: str, message_type: str, content: Dict[str, Any], priority: int = 1):
        message = AgentMessage(
            sender=self.agent_id,
            recipient=recipient,
            message_type=message_type,
            content=content,
            priority=priority
        )

        print(f"📤 {self.agent_id} -> {recipient}: {message_type}")
    
    def receive_message(self, message: AgentMessage):
        self.message_queue.append(message)
        print(f"📥 {self.agent_id} received: {message.message_type} from {message.sender}")
    
    def get_status(self) -> Dict[str, Any]:
        return {
            "agent_id": self.agent_id,
            "agent_type": self.agent_type.value,
            "status": self.status.value,
            "capabilities": self.capabilities,
            "specializations": self.specializations,
            "queue_size": len(self.message_queue)
        }

class PatternAnalystAgent(BaseTTTAgent):
    
    def __init__(self, agent_id: str = "pattern_analyst"):
        super().__init__(agent_id, AgentType.PATTERN_ANALYST)
    
    def _initialize_agent(self):
        self.capabilities = [
            "pattern_recognition",
            "rule_extraction",
            "complexity_assessment",
            "edge_case_identification"
        ]
        self.specializations = [
            "text_patterns",
            "mathematical_patterns",
            "logical_patterns"
        ]
    
    async def process_task(self, context: TaskContext) -> Dict[str, Any]:
        self.status = AgentStatus.WORKING
        
        print(f"🔍 {self.agent_id} analyzing patterns...")
        

        examples = context.examples
        pattern_analysis = {
            "rule_hypothesis": "",
            "confidence_score": 0.0,
            "complexity_level": 0,
            "edge_cases": [],
            "pattern_type": ""
        }
        
        if examples:

            first_example = examples[0]
            input_text = first_example["input"]
            output_text = first_example["output"]
            

            if len(input_text) > 0 and len(output_text) == len(input_text) + 1:
                reversed_part = input_text[::-1]
                if output_text.startswith(reversed_part) and output_text[-1] == input_text[-1]:
                    pattern_analysis["rule_hypothesis"] = "reverse_input_and_append_last_char"
                    pattern_analysis["confidence_score"] = 0.9
                    pattern_analysis["pattern_type"] = "text_transformation"
            

            pattern_analysis["complexity_level"] = min(len(input_text), 5)
            

            pattern_analysis["edge_cases"] = [
                "single_character_input",
                "empty_input",
                "special_characters",
                "numbers_in_input"
            ]
        

        context.results["pattern_analysis"] = pattern_analysis
        context.current_step = "pattern_analyzed"
        
        self.status = AgentStatus.COMPLETED
        

        self.send_message(
            "transformation_expert",
            "pattern_analysis_complete",
            pattern_analysis,
            priority=2
        )
        
        return pattern_analysis

class TransformationExpertAgent(BaseTTTAgent):
    
    def __init__(self, agent_id: str = "transformation_expert"):
        super().__init__(agent_id, AgentType.TRANSFORMATION_EXPERT)
    
    def _initialize_agent(self):
        self.capabilities = [
            "text_transformation",
            "rule_application",
            "step_by_step_execution",
            "error_detection"
        ]
        self.specializations = [
            "string_manipulation",
            "mathematical_transformations",
            "logical_operations"
        ]
    
    async def process_task(self, context: TaskContext) -> Dict[str, Any]:
        self.status = AgentStatus.WORKING
        
        print(f"⚡ {self.agent_id} applying transformation...")
        
        query = context.query
        pattern_analysis = context.results.get("pattern_analysis", {})
        
        transformation_result = {
            "input": query,
            "output": "",
            "steps": [],
            "confidence": 0.0,
            "errors": []
        }
        
        if pattern_analysis.get("rule_hypothesis") == "reverse_input_and_append_last_char":
            try:

                step1 = f"Reverse '{query}' = '{query[::-1]}'"
                step2 = f"Get last character = '{query[-1]}'"
                step3 = f"Append last char = '{query[::-1] + query[-1]}'"
                
                transformation_result["steps"] = [step1, step2, step3]
                transformation_result["output"] = query[::-1] + query[-1]
                transformation_result["confidence"] = pattern_analysis.get("confidence_score", 0.0)
                
            except Exception as e:
                transformation_result["errors"].append(str(e))
                transformation_result["confidence"] = 0.0
        

        context.results["transformation_result"] = transformation_result
        context.current_step = "transformation_applied"
        
        self.status = AgentStatus.COMPLETED
        

        self.send_message(
            "verification_specialist",
            "transformation_complete",
            transformation_result,
            priority=2
        )
        
        return transformation_result

class VerificationSpecialistAgent(BaseTTTAgent):
    
    def __init__(self, agent_id: str = "verification_specialist"):
        super().__init__(agent_id, AgentType.VERIFICATION_SPECIALIST)
    
    def _initialize_agent(self):
        self.capabilities = [
            "result_verification",
            "quality_assessment",
            "consistency_checking",
            "confidence_scoring"
        ]
        self.specializations = [
            "pattern_consistency",
            "output_validation",
            "error_detection"
        ]
    
    async def process_task(self, context: TaskContext) -> Dict[str, Any]:
        self.status = AgentStatus.WORKING
        
        print(f"✅ {self.agent_id} verifying result...")
        
        transformation_result = context.results.get("transformation_result", {})
        examples = context.examples
        
        verification_result = {
            "is_valid": False,
            "confidence_score": 0.0,
            "verification_methods": [],
            "issues": [],
            "quality_score": 0.0
        }
        

        if examples and transformation_result.get("output"):
            output = transformation_result["output"]
            

            for example in examples:
                expected_pattern = example["input"][::-1] + example["input"][-1]
                if expected_pattern == example["output"]:
                    verification_result["verification_methods"].append("pattern_consistency")
                    verification_result["confidence_score"] += 0.3
            

            input_text = transformation_result.get("input", "")
            if len(output) == len(input_text) + 1:
                verification_result["verification_methods"].append("length_check")
                verification_result["confidence_score"] += 0.2
            

            if output == input_text[::-1] + input_text[-1]:
                verification_result["verification_methods"].append("pattern_match")
                verification_result["confidence_score"] += 0.5
                verification_result["is_valid"] = True
        
        verification_result["quality_score"] = min(verification_result["confidence_score"], 1.0)
        

        context.results["verification_result"] = verification_result
        context.current_step = "verification_complete"
        
        self.status = AgentStatus.COMPLETED
        
        return verification_result

class ErrorHandlerAgent(BaseTTTAgent):
    
    def __init__(self, agent_id: str = "error_handler"):
        super().__init__(agent_id, AgentType.ERROR_HANDLER)
    
    def _initialize_agent(self):
        self.capabilities = [
            "error_detection",
            "error_recovery",
            "fallback_strategies",
            "debugging_assistance"
        ]
        self.specializations = [
            "pattern_errors",
            "transformation_errors",
            "verification_errors"
        ]
    
    async def process_task(self, context: TaskContext) -> Dict[str, Any]:
        self.status = AgentStatus.WORKING
        
        print(f"🔧 {self.agent_id} handling errors...")
        
        error_handling_result = {
            "errors_detected": [],
            "recovery_actions": [],
            "fallback_applied": False,
            "success": False
        }
        

        if "pattern_analysis" in context.results:
            pattern_analysis = context.results["pattern_analysis"]
            if pattern_analysis.get("confidence_score", 0) < 0.5:
                error_handling_result["errors_detected"].append("low_pattern_confidence")
                error_handling_result["recovery_actions"].append("request_more_examples")
        
        if "transformation_result" in context.results:
            transformation_result = context.results["transformation_result"]
            if transformation_result.get("errors"):
                error_handling_result["errors_detected"].extend(transformation_result["errors"])
                error_handling_result["recovery_actions"].append("retry_transformation")
        
        if "verification_result" in context.results:
            verification_result = context.results["verification_result"]
            if not verification_result.get("is_valid", False):
                error_handling_result["errors_detected"].append("verification_failed")
                error_handling_result["recovery_actions"].append("apply_fallback_strategy")
                error_handling_result["fallback_applied"] = True
        

        if error_handling_result["recovery_actions"]:
            error_handling_result["success"] = True
        
        self.status = AgentStatus.COMPLETED
        
        return error_handling_result

class CoordinatorAgent(BaseTTTAgent):
    
    def __init__(self, agent_id: str = "coordinator"):
        super().__init__(agent_id, AgentType.COORDINATOR)
        self.subordinate_agents = {}
    
    def _initialize_agent(self):
        self.capabilities = [
            "workflow_coordination",
            "agent_management",
            "task_scheduling",
            "result_aggregation"
        ]
        self.specializations = [
            "multi_agent_coordination",
            "workflow_optimization",
            "resource_management"
        ]
    
    def register_agent(self, agent: BaseTTTAgent):
        self.subordinate_agents[agent.agent_id] = agent
        print(f"📝 Registered agent: {agent.agent_id}")
    
    async def process_task(self, context: TaskContext) -> Dict[str, Any]:
        self.status = AgentStatus.WORKING
        
        print(f"🎯 {self.agent_id} coordinating TTT process...")
        
        coordination_result = {
            "workflow_steps": [],
            "agent_performance": {},
            "overall_success": False,
            "final_result": "",
            "total_time": 0.0
        }
        
        start_time = time.time()
        
        try:

            if "pattern_analyst" in self.subordinate_agents:
                pattern_agent = self.subordinate_agents["pattern_analyst"]
                pattern_result = await pattern_agent.process_task(context)
                coordination_result["workflow_steps"].append("pattern_analysis")
                coordination_result["agent_performance"]["pattern_analyst"] = pattern_result
            

            if "transformation_expert" in self.subordinate_agents:
                transform_agent = self.subordinate_agents["transformation_expert"]
                transform_result = await transform_agent.process_task(context)
                coordination_result["workflow_steps"].append("transformation")
                coordination_result["agent_performance"]["transformation_expert"] = transform_result
            

            if "verification_specialist" in self.subordinate_agents:
                verify_agent = self.subordinate_agents["verification_specialist"]
                verify_result = await verify_agent.process_task(context)
                coordination_result["workflow_steps"].append("verification")
                coordination_result["agent_performance"]["verification_specialist"] = verify_result
            

            if "error_handler" in self.subordinate_agents:
                error_agent = self.subordinate_agents["error_handler"]
                error_result = await error_agent.process_task(context)
                coordination_result["workflow_steps"].append("error_handling")
                coordination_result["agent_performance"]["error_handler"] = error_result
            

            transformation_result = context.results.get("transformation_result", {})
            coordination_result["final_result"] = transformation_result.get("output", "")
            coordination_result["overall_success"] = bool(coordination_result["final_result"])
            
        except Exception as e:
            coordination_result["workflow_steps"].append("error")
            coordination_result["agent_performance"]["error"] = str(e)
        
        coordination_result["total_time"] = time.time() - start_time
        
        self.status = AgentStatus.COMPLETED
        
        return coordination_result

class CreativeGeneratorAgent(BaseTTTAgent):
    
    def __init__(self, agent_id: str = "creative_generator"):
        super().__init__(agent_id, AgentType.CREATIVE_GENERATOR)
    
    def _initialize_agent(self):
        self.capabilities = [
            "creative_example_generation",
            "task_variation_creation",
            "novel_pattern_discovery",
            "inspiration_providing"
        ]
        self.specializations = [
            "text_creativity",
            "pattern_innovation",
            "example_diversification"
        ]
    
    async def process_task(self, context: TaskContext) -> Dict[str, Any]:
        self.status = AgentStatus.WORKING
        
        print(f"🎨 {self.agent_id} generating creative content...")
        
        creative_result = {
            "creative_examples": [],
            "task_variations": [],
            "novel_patterns": [],
            "inspiration_score": 0.0
        }
        

        base_examples = context.examples
        if base_examples:

            for example in base_examples[:2]:
                input_text = example["input"]
                output_text = example["output"]
                

                creative_examples = [
                    {"input": input_text.upper(), "output": output_text.upper()},
                    {"input": input_text.lower(), "output": output_text.lower()},
                    {"input": input_text[::-1], "output": output_text[::-1]}
                ]
                
                creative_result["creative_examples"].extend(creative_examples)
        

        task_description = context.task_description
        creative_result["task_variations"] = [
            f"{task_description} with uppercase letters",
            f"{task_description} with numbers",
            f"{task_description} with special characters"
        ]
        

        creative_result["novel_patterns"] = [
            "reverse_and_capitalize_first",
            "reverse_and_add_random_char",
            "reverse_and_duplicate_last"
        ]
        
        creative_result["inspiration_score"] = 0.8
        
        self.status = AgentStatus.COMPLETED
        
        return creative_result

class MultiAgentTTTSystem:
    
    def __init__(self):
        self.agents = {}
        self.coordinator = None
        self.message_broker = []
        self._initialize_system()
    
    def _initialize_system(self):
        print("🤖 Initializing Multi-Agent TTT System...")
        

        self.agents["pattern_analyst"] = PatternAnalystAgent()
        self.agents["transformation_expert"] = TransformationExpertAgent()
        self.agents["verification_specialist"] = VerificationSpecialistAgent()
        self.agents["error_handler"] = ErrorHandlerAgent()
        self.agents["creative_generator"] = CreativeGeneratorAgent()
        

        self.coordinator = CoordinatorAgent()
        

        for agent in self.agents.values():
            self.coordinator.register_agent(agent)
        
        print("✅ Multi-Agent TTT System initialized!")
    
    async def run_ttt_task(self, task_description: str, examples: List[Dict[str, str]], 
                          query: str) -> Dict[str, Any]:
        print(f"🚀 Running TTT task with multi-agent system...")
        print(f"📋 Task: {task_description}")
        print(f"🔍 Query: {query}")
        

        context = TaskContext(
            task_id=f"task_{int(time.time())}",
            task_description=task_description,
            examples=examples,
            query=query,
            current_step="initialized"
        )
        

        result = await self.coordinator.process_task(context)
        

        result["context_results"] = context.results
        result["task_context"] = {
            "task_id": context.task_id,
            "task_description": context.task_description,
            "query": context.query,
            "current_step": context.current_step
        }
        
        return result
    
    def get_system_status(self) -> Dict[str, Any]:
        status = {
            "coordinator": self.coordinator.get_status() if self.coordinator else None,
            "agents": {}
        }
        
        for agent_id, agent in self.agents.items():
            status["agents"][agent_id] = agent.get_status()
        
        return status


async def test_multi_agent_system():
    print("🤖 Testing Multi-Agent TTT System")
    print("=" * 50)
    

    system = MultiAgentTTTSystem()
    

    task_description = "Reverse the word and append the last character"
    examples = [
        {"input": "hello", "output": "olleho"},
        {"input": "world", "output": "dlrowd"},
        {"input": "python", "output": "nohtypp"}
    ]
    test_queries = ["gemini", "tensor", "neural"]
    
    print(f"📋 Task: {task_description}")
    print(f"📝 Examples: {len(examples)}")
    print(f"🔍 Test Queries: {test_queries}")
    

    results = []
    for query in test_queries:
        print(f"\n🔍 Testing query: {query}")
        
        result = await system.run_ttt_task(task_description, examples, query)
        results.append(result)
        
        print(f"   ✅ Result: {result.get('final_result', 'N/A')}")
        print(f"   📊 Success: {result.get('overall_success', False)}")
        print(f"   ⏱️ Time: {result.get('total_time', 0):.3f}s")
    

    print(f"\n📊 System Status:")
    status = system.get_system_status()
    for agent_id, agent_status in status["agents"].items():
        print(f"   {agent_id}: {agent_status['status']}")
    
    print("✅ Multi-agent system test completed!")
    return results

if __name__ == "__main__":

    asyncio.run(test_multi_agent_system())

Xet Storage Details

Size:: 21.5 kB
Xet hash:: 9745288a515c6a919517f4d796cbc8ba1c25c9b61622d07c604b71cb7f70d8b3

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

	import os
	import json
	import asyncio
	from typing import List, Dict, Any, Optional, Union, Callable
	from dataclasses import dataclass, field
	from enum import Enum
	from abc import ABC, abstractmethod
	import time

	try:
	from langchain.agents import AgentExecutor, create_react_agent
	from langchain.tools import BaseTool, Tool
	from langchain.schema import BaseMessage, HumanMessage, AIMessage, SystemMessage
	from langchain.llms import OpenAI
	from langchain.chat_models import ChatOpenAI
	LANCHAIN_AVAILABLE = True
	except ImportError:
	LANCHAIN_AVAILABLE = False

	try:
	import google.generativeai as genai
	GEMINI_AVAILABLE = True
	except ImportError:
	GEMINI_AVAILABLE = False

	class AgentType(Enum):
	PATTERN_ANALYST = "pattern_analyst"
	TRANSFORMATION_EXPERT = "transformation_expert"
	VERIFICATION_SPECIALIST = "verification_specialist"
	ERROR_HANDLER = "error_handler"
	COORDINATOR = "coordinator"
	CREATIVE_GENERATOR = "creative_generator"

	class AgentStatus(Enum):
	IDLE = "idle"
	WORKING = "working"
	WAITING = "waiting"
	ERROR = "error"
	COMPLETED = "completed"

	@dataclass
	class AgentMessage:
	sender: str
	recipient: str
	message_type: str
	content: Dict[str, Any]
	timestamp: float = field(default_factory=time.time)
	priority: int = 1

	@dataclass
	class TaskContext:
	task_id: str
	task_description: str
	examples: List[Dict[str, str]]
	query: str
	current_step: str
	results: Dict[str, Any] = field(default_factory=dict)
	metadata: Dict[str, Any] = field(default_factory=dict)

	class BaseTTTAgent(ABC):

	def __init__(self, agent_id: str, agent_type: AgentType):
	self.agent_id = agent_id
	self.agent_type = agent_type
	self.status = AgentStatus.IDLE
	self.message_queue = []
	self.capabilities = []
	self.specializations = []
	self._initialize_agent()

	@abstractmethod
	def _initialize_agent(self):
	pass

	@abstractmethod
	async def process_task(self, context: TaskContext) -> Dict[str, Any]:
	pass

	def send_message(self, recipient: str, message_type: str, content: Dict[str, Any], priority: int = 1):
	message = AgentMessage(
	sender=self.agent_id,
	recipient=recipient,
	message_type=message_type,
	content=content,
	priority=priority
	)

	print(f"📤 {self.agent_id} -> {recipient}: {message_type}")

	def receive_message(self, message: AgentMessage):
	self.message_queue.append(message)
	print(f"📥 {self.agent_id} received: {message.message_type} from {message.sender}")

	def get_status(self) -> Dict[str, Any]:
	return {
	"agent_id": self.agent_id,
	"agent_type": self.agent_type.value,
	"status": self.status.value,
	"capabilities": self.capabilities,
	"specializations": self.specializations,
	"queue_size": len(self.message_queue)
	}

	class PatternAnalystAgent(BaseTTTAgent):

	def __init__(self, agent_id: str = "pattern_analyst"):
	super().__init__(agent_id, AgentType.PATTERN_ANALYST)

	def _initialize_agent(self):
	self.capabilities = [
	"pattern_recognition",
	"rule_extraction",
	"complexity_assessment",
	"edge_case_identification"
	]
	self.specializations = [
	"text_patterns",
	"mathematical_patterns",
	"logical_patterns"
	]

	async def process_task(self, context: TaskContext) -> Dict[str, Any]:
	self.status = AgentStatus.WORKING

	print(f"🔍 {self.agent_id} analyzing patterns...")


	examples = context.examples
	pattern_analysis = {
	"rule_hypothesis": "",
	"confidence_score": 0.0,
	"complexity_level": 0,
	"edge_cases": [],
	"pattern_type": ""
	}

	if examples:

	first_example = examples[0]
	input_text = first_example["input"]
	output_text = first_example["output"]


	if len(input_text) > 0 and len(output_text) == len(input_text) + 1:
	reversed_part = input_text[::-1]
	if output_text.startswith(reversed_part) and output_text[-1] == input_text[-1]:
	pattern_analysis["rule_hypothesis"] = "reverse_input_and_append_last_char"
	pattern_analysis["confidence_score"] = 0.9
	pattern_analysis["pattern_type"] = "text_transformation"


	pattern_analysis["complexity_level"] = min(len(input_text), 5)


	pattern_analysis["edge_cases"] = [
	"single_character_input",
	"empty_input",
	"special_characters",
	"numbers_in_input"
	]


	context.results["pattern_analysis"] = pattern_analysis
	context.current_step = "pattern_analyzed"

	self.status = AgentStatus.COMPLETED


	self.send_message(
	"transformation_expert",
	"pattern_analysis_complete",
	pattern_analysis,
	priority=2
	)

	return pattern_analysis

	class TransformationExpertAgent(BaseTTTAgent):

	def __init__(self, agent_id: str = "transformation_expert"):
	super().__init__(agent_id, AgentType.TRANSFORMATION_EXPERT)

	def _initialize_agent(self):
	self.capabilities = [
	"text_transformation",
	"rule_application",
	"step_by_step_execution",
	"error_detection"
	]
	self.specializations = [
	"string_manipulation",
	"mathematical_transformations",
	"logical_operations"
	]

	async def process_task(self, context: TaskContext) -> Dict[str, Any]:
	self.status = AgentStatus.WORKING

	print(f"⚡ {self.agent_id} applying transformation...")

	query = context.query
	pattern_analysis = context.results.get("pattern_analysis", {})

	transformation_result = {
	"input": query,
	"output": "",
	"steps": [],
	"confidence": 0.0,
	"errors": []
	}

	if pattern_analysis.get("rule_hypothesis") == "reverse_input_and_append_last_char":
	try:

	step1 = f"Reverse '{query}' = '{query[::-1]}'"
	step2 = f"Get last character = '{query[-1]}'"
	step3 = f"Append last char = '{query[::-1] + query[-1]}'"

	transformation_result["steps"] = [step1, step2, step3]
	transformation_result["output"] = query[::-1] + query[-1]
	transformation_result["confidence"] = pattern_analysis.get("confidence_score", 0.0)

	except Exception as e:
	transformation_result["errors"].append(str(e))
	transformation_result["confidence"] = 0.0


	context.results["transformation_result"] = transformation_result
	context.current_step = "transformation_applied"

	self.status = AgentStatus.COMPLETED


	self.send_message(
	"verification_specialist",
	"transformation_complete",
	transformation_result,
	priority=2
	)

	return transformation_result

	class VerificationSpecialistAgent(BaseTTTAgent):

	def __init__(self, agent_id: str = "verification_specialist"):
	super().__init__(agent_id, AgentType.VERIFICATION_SPECIALIST)

	def _initialize_agent(self):
	self.capabilities = [
	"result_verification",
	"quality_assessment",
	"consistency_checking",
	"confidence_scoring"
	]
	self.specializations = [
	"pattern_consistency",
	"output_validation",
	"error_detection"
	]

	async def process_task(self, context: TaskContext) -> Dict[str, Any]:
	self.status = AgentStatus.WORKING

	print(f"✅ {self.agent_id} verifying result...")

	transformation_result = context.results.get("transformation_result", {})
	examples = context.examples

	verification_result = {
	"is_valid": False,
	"confidence_score": 0.0,
	"verification_methods": [],
	"issues": [],
	"quality_score": 0.0
	}


	if examples and transformation_result.get("output"):
	output = transformation_result["output"]


	for example in examples:
	expected_pattern = example["input"][::-1] + example["input"][-1]
	if expected_pattern == example["output"]:
	verification_result["verification_methods"].append("pattern_consistency")
	verification_result["confidence_score"] += 0.3


	input_text = transformation_result.get("input", "")
	if len(output) == len(input_text) + 1:
	verification_result["verification_methods"].append("length_check")
	verification_result["confidence_score"] += 0.2


	if output == input_text[::-1] + input_text[-1]:
	verification_result["verification_methods"].append("pattern_match")
	verification_result["confidence_score"] += 0.5
	verification_result["is_valid"] = True

	verification_result["quality_score"] = min(verification_result["confidence_score"], 1.0)


	context.results["verification_result"] = verification_result
	context.current_step = "verification_complete"

	self.status = AgentStatus.COMPLETED

	return verification_result

	class ErrorHandlerAgent(BaseTTTAgent):

	def __init__(self, agent_id: str = "error_handler"):
	super().__init__(agent_id, AgentType.ERROR_HANDLER)

	def _initialize_agent(self):
	self.capabilities = [
	"error_detection",
	"error_recovery",
	"fallback_strategies",
	"debugging_assistance"
	]
	self.specializations = [
	"pattern_errors",
	"transformation_errors",
	"verification_errors"
	]

	async def process_task(self, context: TaskContext) -> Dict[str, Any]:
	self.status = AgentStatus.WORKING

	print(f"🔧 {self.agent_id} handling errors...")

	error_handling_result = {
	"errors_detected": [],
	"recovery_actions": [],
	"fallback_applied": False,
	"success": False
	}


	if "pattern_analysis" in context.results:
	pattern_analysis = context.results["pattern_analysis"]
	if pattern_analysis.get("confidence_score", 0) < 0.5:
	error_handling_result["errors_detected"].append("low_pattern_confidence")
	error_handling_result["recovery_actions"].append("request_more_examples")

	if "transformation_result" in context.results:
	transformation_result = context.results["transformation_result"]
	if transformation_result.get("errors"):
	error_handling_result["errors_detected"].extend(transformation_result["errors"])
	error_handling_result["recovery_actions"].append("retry_transformation")

	if "verification_result" in context.results:
	verification_result = context.results["verification_result"]
	if not verification_result.get("is_valid", False):
	error_handling_result["errors_detected"].append("verification_failed")
	error_handling_result["recovery_actions"].append("apply_fallback_strategy")
	error_handling_result["fallback_applied"] = True


	if error_handling_result["recovery_actions"]:
	error_handling_result["success"] = True

	self.status = AgentStatus.COMPLETED

	return error_handling_result

	class CoordinatorAgent(BaseTTTAgent):

	def __init__(self, agent_id: str = "coordinator"):
	super().__init__(agent_id, AgentType.COORDINATOR)
	self.subordinate_agents = {}

	def _initialize_agent(self):
	self.capabilities = [
	"workflow_coordination",
	"agent_management",
	"task_scheduling",
	"result_aggregation"
	]
	self.specializations = [
	"multi_agent_coordination",
	"workflow_optimization",
	"resource_management"
	]

	def register_agent(self, agent: BaseTTTAgent):
	self.subordinate_agents[agent.agent_id] = agent
	print(f"📝 Registered agent: {agent.agent_id}")

	async def process_task(self, context: TaskContext) -> Dict[str, Any]:
	self.status = AgentStatus.WORKING

	print(f"🎯 {self.agent_id} coordinating TTT process...")

	coordination_result = {
	"workflow_steps": [],
	"agent_performance": {},
	"overall_success": False,
	"final_result": "",
	"total_time": 0.0
	}

	start_time = time.time()

	try:

	if "pattern_analyst" in self.subordinate_agents:
	pattern_agent = self.subordinate_agents["pattern_analyst"]
	pattern_result = await pattern_agent.process_task(context)
	coordination_result["workflow_steps"].append("pattern_analysis")
	coordination_result["agent_performance"]["pattern_analyst"] = pattern_result


	if "transformation_expert" in self.subordinate_agents:
	transform_agent = self.subordinate_agents["transformation_expert"]
	transform_result = await transform_agent.process_task(context)
	coordination_result["workflow_steps"].append("transformation")
	coordination_result["agent_performance"]["transformation_expert"] = transform_result


	if "verification_specialist" in self.subordinate_agents:
	verify_agent = self.subordinate_agents["verification_specialist"]
	verify_result = await verify_agent.process_task(context)
	coordination_result["workflow_steps"].append("verification")
	coordination_result["agent_performance"]["verification_specialist"] = verify_result


	if "error_handler" in self.subordinate_agents:
	error_agent = self.subordinate_agents["error_handler"]
	error_result = await error_agent.process_task(context)
	coordination_result["workflow_steps"].append("error_handling")
	coordination_result["agent_performance"]["error_handler"] = error_result


	transformation_result = context.results.get("transformation_result", {})
	coordination_result["final_result"] = transformation_result.get("output", "")
	coordination_result["overall_success"] = bool(coordination_result["final_result"])

	except Exception as e:
	coordination_result["workflow_steps"].append("error")
	coordination_result["agent_performance"]["error"] = str(e)

	coordination_result["total_time"] = time.time() - start_time

	self.status = AgentStatus.COMPLETED

	return coordination_result

	class CreativeGeneratorAgent(BaseTTTAgent):

	def __init__(self, agent_id: str = "creative_generator"):
	super().__init__(agent_id, AgentType.CREATIVE_GENERATOR)

	def _initialize_agent(self):
	self.capabilities = [
	"creative_example_generation",
	"task_variation_creation",
	"novel_pattern_discovery",
	"inspiration_providing"
	]
	self.specializations = [
	"text_creativity",
	"pattern_innovation",
	"example_diversification"
	]

	async def process_task(self, context: TaskContext) -> Dict[str, Any]:
	self.status = AgentStatus.WORKING

	print(f"🎨 {self.agent_id} generating creative content...")

	creative_result = {
	"creative_examples": [],
	"task_variations": [],
	"novel_patterns": [],
	"inspiration_score": 0.0
	}


	base_examples = context.examples
	if base_examples:

	for example in base_examples[:2]:
	input_text = example["input"]
	output_text = example["output"]


	creative_examples = [
	{"input": input_text.upper(), "output": output_text.upper()},
	{"input": input_text.lower(), "output": output_text.lower()},
	{"input": input_text[::-1], "output": output_text[::-1]}
	]

	creative_result["creative_examples"].extend(creative_examples)


	task_description = context.task_description
	creative_result["task_variations"] = [
	f"{task_description} with uppercase letters",
	f"{task_description} with numbers",
	f"{task_description} with special characters"
	]


	creative_result["novel_patterns"] = [
	"reverse_and_capitalize_first",
	"reverse_and_add_random_char",
	"reverse_and_duplicate_last"
	]

	creative_result["inspiration_score"] = 0.8

	self.status = AgentStatus.COMPLETED

	return creative_result

	class MultiAgentTTTSystem:

	def __init__(self):
	self.agents = {}
	self.coordinator = None
	self.message_broker = []
	self._initialize_system()

	def _initialize_system(self):
	print("🤖 Initializing Multi-Agent TTT System...")


	self.agents["pattern_analyst"] = PatternAnalystAgent()
	self.agents["transformation_expert"] = TransformationExpertAgent()
	self.agents["verification_specialist"] = VerificationSpecialistAgent()
	self.agents["error_handler"] = ErrorHandlerAgent()
	self.agents["creative_generator"] = CreativeGeneratorAgent()


	self.coordinator = CoordinatorAgent()


	for agent in self.agents.values():
	self.coordinator.register_agent(agent)

	print("✅ Multi-Agent TTT System initialized!")

	async def run_ttt_task(self, task_description: str, examples: List[Dict[str, str]],
	query: str) -> Dict[str, Any]:
	print(f"🚀 Running TTT task with multi-agent system...")
	print(f"📋 Task: {task_description}")
	print(f"🔍 Query: {query}")


	context = TaskContext(
	task_id=f"task_{int(time.time())}",
	task_description=task_description,
	examples=examples,
	query=query,
	current_step="initialized"
	)


	result = await self.coordinator.process_task(context)


	result["context_results"] = context.results
	result["task_context"] = {
	"task_id": context.task_id,
	"task_description": context.task_description,
	"query": context.query,
	"current_step": context.current_step
	}

	return result

	def get_system_status(self) -> Dict[str, Any]:
	status = {
	"coordinator": self.coordinator.get_status() if self.coordinator else None,
	"agents": {}
	}

	for agent_id, agent in self.agents.items():
	status["agents"][agent_id] = agent.get_status()

	return status


	async def test_multi_agent_system():
	print("🤖 Testing Multi-Agent TTT System")
	print("=" * 50)


	system = MultiAgentTTTSystem()


	task_description = "Reverse the word and append the last character"
	examples = [
	{"input": "hello", "output": "olleho"},
	{"input": "world", "output": "dlrowd"},
	{"input": "python", "output": "nohtypp"}
	]
	test_queries = ["gemini", "tensor", "neural"]

	print(f"📋 Task: {task_description}")
	print(f"📝 Examples: {len(examples)}")
	print(f"🔍 Test Queries: {test_queries}")


	results = []
	for query in test_queries:
	print(f"\n🔍 Testing query: {query}")

	result = await system.run_ttt_task(task_description, examples, query)
	results.append(result)

	print(f" ✅ Result: {result.get('final_result', 'N/A')}")
	print(f" 📊 Success: {result.get('overall_success', False)}")
	print(f" ⏱️ Time: {result.get('total_time', 0):.3f}s")


	print(f"\n📊 System Status:")
	status = system.get_system_status()
	for agent_id, agent_status in status["agents"].items():
	print(f" {agent_id}: {agent_status['status']}")

	print("✅ Multi-agent system test completed!")
	return results

	if __name__ == "__main__":

	asyncio.run(test_multi_agent_system())