from typing import List, Dict
from dataclasses import dataclass

@dataclass
class Agent:
    id: str
    role: str
    capabilities: List[str]

@dataclass
class Task:
    id: str
    requirements: List[str]
    priority: int

class HierarchicalOrchestrator:
    def __init__(self):
        self.agents = []
        self.task_queue = []
    
    def assign_task(self, task: Task):
        # Find best agent based on capabilities and load
        suitable_agents = [
            agent for agent in self.agents 
            if all(req in agent.capabilities for req in task.requirements)
        ]
        
        if suitable_agents:
            # Simple round-robin assignment
            selected = min(suitable_agents, key=lambda a: a.current_load)
            selected.current_load += 1
            return selected.id
        return None
    
    def orchestrate(self, tasks: List[Task]):
        # Central coordinator logic
        for task in sorted(tasks, key=lambda t: t.priority, reverse=True):
            agent_id = self.assign_task(task)
            if agent_id:
                print(f"Assigned task {task.id} to agent {agent_id}")

import asyncio
from typing import Dict, List

class DecentralizedAgent:
    def __init__(self, id: str, capabilities: List[str]):
        self.id = id
        self.capabilities = capabilities
        self.available = True
        self.task_queue = []
    
    async def negotiate_task(self, task: Dict, other_agents: List['DecentralizedAgent']):
        # Auction-based negotiation
        bid = self.estimate_cost(task)
        bids = await asyncio.gather(
            *(agent.bid_for_task(task) for agent in other_agents if agent != self)
        )
        
        if bid == min([bid] + bids):
            return True  # Won the auction
        return False
    
    def bid_for_task(self, task: Dict):
        # Simple capability matching bid
        if all(req in self.capabilities for req in task['requirements']):
            return len(task['requirements'])  # Lower bid for better match
        return float('inf')  # Can't handle task
    
    def execute_task(self, task: Dict):
        print(f"Agent {self.id} executing task: {task['id']}")
        # Task execution logic

import requests

class AgentCommunicator:
    def send_request(self, agent_url: str, payload: dict) -> dict:
        response = requests.post(agent_url, json=payload)
        return response.json()
    
    def receive_request(self, request_data: dict) -> dict:
        # Process incoming request
        return {"status": "accepted", "agent_id": self.id}

from typing import Callable, List
import asyncio

class MessageBus:
    def __init__(self):
        self.subscribers: Dict[str, List[Callable]] = {}
    
    def subscribe(self, topic: str, callback: Callable):
        if topic not in self.subscribers:
            self.subscribers[topic] = []
        self.subscribers[topic].append(callback)
    
    async def publish(self, topic: str, message: dict):
        if topic in self.subscribers:
            for callback in self.subscribers[topic]:
                await callback(message)

from langchain.agents import create_openai_functions_agent
from langchain.tools import Tool
from langchain.graph import StateGraph

# Define specialized tools
tools = [
    Tool(
        name="search_web",
        func=lambda query: f"Search results for: {query}",
        description="Search the web for information"
    ),
    Tool(
        name="analyze_data",
        func=lambda data: f"Analysis results: {data}",
        description="Analyze provided data"
    )
]

# Create multi-agent system
graph = StateGraph()

# Define agent states and transitions
graph.add_state("initial", "initial_state")
graph.add_state("research", "research_state")
graph.add_state("analysis", "analysis_state")
graph.add_state("final", "final_state")

# Define transitions
graph.add_transition("initial", "research", "needs_research")
graph.add_transition("research", "analysis", "research_complete")
graph.add_transition("analysis", "final", "analysis_complete")

# Create agent with state management
agent = create_openai_functions_agent(tools, prompt)

from crewai import Agent, Task, Process, Crew

# Define specialized agents
researcher = Agent(
    role="Researcher",
    goal="Gather comprehensive information",
    backstory="Experienced research analyst with 10 years in data gathering",
    tools=["web_search", "data_analysis"]
)

analyst = Agent(
    role="Data Analyst",
    goal="Transform raw data into actionable insights",
    backstory="Skilled statistician specializing in pattern recognition",
    tools=["statistical_analysis", "visualization"]
)

writer = Agent(
    role="Technical Writer",
    goal="Communicate findings clearly and accurately",
    backstory="Professional writer with expertise in technical documentation",
    tools=["content_creation", "editing"]
)

# Define tasks
research_task = Task(
    description="Research current AI trends and market data",
    agent=researcher,
    expected_output="Comprehensive research report"
)

analysis_task = Task(
    description="Analyze research data for key patterns",
    agent=analyst,
    expected_output="Data analysis with visualizations"
)

writing_task = Task(
    description="Create a comprehensive report from analysis",
    agent=writer,
    expected_output="Final report document"
)

# Create crew with process
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.DIRECTED_CYCLIC_GRAPH
)

# Execute
crew.kickoff()

from autogen import ChatBot, TaskBot, Conversation

# Create specialized bots
research_bot = ChatBot(
    name="Researcher",
    system_message="You are an expert researcher who can search the web and analyze data.",
    capabilities=["search", "analyze"]
)

analysis_bot = ChatBot(
    name="Analyst",
    system_message="You are a data analyst who can interpret research findings.",
    capabilities=["interpret", "visualize"]
)

# Create conversation with multi-agent setup
conversation = Conversation(
    bots=[research_bot, analysis_bot],
    human_input=[
        "Research current trends in AI agent frameworks",
        "Analyze the key findings and create a summary"
    ]
)

# Run conversation with automatic agent selection
conversation.run()

from typing import List, Dict
import heapq

class GoalPlanner:
    def __init__(self, agents: List[Dict]):
        self.agents = agents
    
    def plan_goal(self, goal: str, context: Dict) -> List[Dict]:
        # Hierarchical task network planning
        if goal == "research_market_trends":
            return [
                {"task": "identify_sources", "priority": 1},
                {"task": "gather_data", "priority": 2},
                {"task": "analyze_data", "priority": 3},
                {"task": "generate_report", "priority": 4}
            ]
        
        return []
    
    def execute_plan(self, plan: List[Dict], context: Dict):
        # Priority-based execution using heap
        heapq.heapify(plan)
        
        while plan:
            current_task = heapq.heappop(plan)
            agent = self.select_agent(current_task['task'])
            if agent:
                agent.execute_task(current_task['task'], context)
    
    def select_agent(self, task_type: str) -> 'Agent':
        # Capability-based agent selection
        for agent in self.agents:
            if task_type in agent.capabilities:
                return agent
        return None

import random
import time

class AdaptivePlanner:
    def __init__(self, agents: List[Dict]):
        self.agents = agents
        self.plan = []
        self.context = {}
    
    def monitor_execution(self):
        # Simulate monitoring with random failures
        while self.plan:
            task = self.plan[0]
            success = self.execute_task_with_retry(task)
            
            if not success:
                print(f"Task {task['task']} failed. Replanning...")
                self.replan(task)
            else:
                self.plan.pop(0)
    
    def execute_task_with_retry(self, task: Dict, max_retries=3) -> bool:
        for attempt in range(max_retries):
            agent = self.select_agent(task['task'])
            if agent:
                try:
                    agent.execute_task(task['task'], self.context)
                    return True
                except Exception as e:
                    print(f"Attempt {attempt + 1} failed: {e}")
                    time.sleep(2 ** attempt)  # Exponential backoff
        return False
    
    def replan(self, failed_task: Dict):
        # Simple replanning: retry with different agent or approach
        alternative_agents = [
            agent for agent in self.agents 
            if agent.id != failed_task['assigned_agent']
        ]
        
        if alternative_agents:
            new_agent = random.choice(alternative_agents)
            self.plan.insert(0, {
                **failed_task,
                'assigned_agent': new_agent.id,
                'attempt': failed_task.get('attempt', 0) + 1
            })

from crewai import Agent, Task, Crew

# Support crew agents
classifier = Agent(
    role="Ticket Classifier",
    goal="Categorize incoming support tickets",
    tools=["text_classification", "intent_detection"]
)

researcher = Agent(
    role="Knowledge Researcher",
    goal="Find relevant solutions from knowledge base",
    tools=["kb_search", "document_retrieval"]
)

resolver = Agent(
    role="Problem Resolver",
    goal="Provide accurate solutions to customer issues",
    tools=["solution_generation", "troubleshooting"]
)

# Support tasks
classify_task = Task(
    description="Categorize and prioritize incoming ticket",
    agent=classifier
)

research_task = Task(
    description="Find relevant solutions from knowledge base",
    agent=researcher
)

resolve_task = Task(
    description="Generate solution and response to customer",
    agent=resolver
)

# Support crew
support_crew = Crew(
    agents=[classifier, researcher, resolver],
    tasks=[classify_task, research_task, resolve_task],
    process=Process.SEQUENTIAL
)

# Execute support workflow
def handle_support_ticket(ticket: dict):
    classify_task.input = ticket
    research_task.input = ticket
    resolve_task.input = ticket
    
    support_crew.kickoff()
    return resolve_task.output

from typing import Dict, List
import pandas as pd

class FinancialAnalysisAgent:
    def __init__(self, role: str, specialty: str):
        self.role = role
        self.specialty = specialty
    
    def analyze(self, data: pd.DataFrame) -> Dict:
        if self.specialty == "market_trends":
            return self.analyze_market_trends(data)
        elif self.specialty == "risk_assessment":
            return self.assess_risks(data)
        elif self.specialty == "valuation":
            return self.calculate_valuations(data)
    
    def analyze_market_trends(self, data: pd.DataFrame) -> Dict:
        # Technical analysis
        trends = data.rolling(window=20).mean()
        return {"trends": trends, "volatility": trends.std()}
    
    def assess_risks(self, data: pd.DataFrame) -> Dict:
        # Risk metrics
        risk_metrics = {
            "var": data.quantile(0.05),
            "sharpe_ratio": data.mean() / data.std()
        }
        return risk_metrics
    
    def calculate_valuations(self, data: pd.DataFrame) -> Dict:
        # Valuation models
        valuations = {
            "dcf_valuation": data.sum() * 1.5,
            "comparable_valuation": data.mean() * 12
        }
        return valuations

# Multi-agent financial analysis
class FinancialAnalysisOrchestrator:
    def __init__(self):
        self.agents = [
            FinancialAnalysisAgent("Market Analyst", "market_trends"),
            FinancialAnalysisAgent("Risk Manager", "risk_assessment"),
            FinancialAnalysisAgent("Valuation Expert", "valuation")
        ]
    
    def generate_report(self, company_data: pd.DataFrame) -> Dict:
        analysis_results = {}
        
        for agent in self.agents:
            results = agent.analyze(company_data)
            analysis_results[agent.role] = results
        
        # Synthesize findings
        final_report = {
            "executive_summary": self.synthesize_findings(analysis_results),
            "detailed_analysis": analysis_results
        }
        
        return final_report
    
    def synthesize_findings(self, analysis_results: Dict) -> str:
        # Simple synthesis logic
        summary = "Our analysis indicates a mixed outlook. Market trends show moderate growth, while risk assessment reveals manageable volatility. Valuation suggests potential upside."
        return summary