Arquitetura multi-agent: orquestrando a complexidade

import json

class MultiAgentOrchestrator:
    def __init__(self):
        self.agents = {
            "database": Agent(
                system_prompt="Especialista em PostgreSQL, MySQL, Cosmos DB.",
                tools=DB_TOOLS,
            ),
            "network": Agent(
                system_prompt="Especialista em DNS, firewall, VPN, load balancers.",
                tools=NET_TOOLS,
            ),
            "kubernetes": Agent(
                system_prompt="Especialista em AKS, pods, deployments, networking K8s.",
                tools=K8S_TOOLS,
            ),
        }
    
    def run(self, task):
        # Orchestrator analisa e decompõe
        plan = self.plan(task)
        
        results = []
        for step in plan["steps"]:
            agent = self.agents[step["assigned_to"]]
            result = agent.execute(step["task"])
            results.append({
                "step": step,
                "result": result,
                "agent": step["assigned_to"],
            })
            
            # Orchestrator pode re-planejar baseado em resultados
            if self.needs_replanning(plan, results):
                plan = self.replan(task, plan, results)
        
        # Consolidar resultados de todos os agents
        return self.synthesize(task, results)
    
    def plan(self, task):
        response = client.chat.completions.create(
            model="gpt-4o",
            temperature=0,
            messages=[
                {"role": "system", "content": """
Você é o orquestrador. Decomponha a tarefa em steps e atribua cada step
ao agent mais adequado.

Agents disponíveis:
- database: problemas com bancos de dados
- network: problemas de rede e conectividade
- kubernetes: problemas com clusters K8s

Retorne JSON válido no formato:
{"steps": [{"task": "...", "assigned_to": "database|network|kubernetes", "depends_on": []}]}
"""},
                {"role": "user", "content": task},
            ],
            response_format={"type": "json_object"},
        )
        return json.loads(response.choices[0].message.content)

def pipeline_agents(alert):
    # Stage 1: coletar informações
    context = collector_agent.run(
        f"Colete métricas, logs e status de: {alert['resource']}"
    )
    
    # Stage 2: analisar
    diagnosis = analyzer_agent.run(
        f"Analise e identifique causa raiz:\nDados: {context}"
    )
    
    # Stage 3: planejar remediação
    plan = planner_agent.run(
        f"Crie plano de remediação:\nDiagnóstico: {diagnosis}"
    )
    
    # Stage 4: executar (com approval se necessário)
    if plan["risk"] == "high":
        await_human_approval(plan)
    
    result = executor_agent.run(
        f"Execute o plano: {plan}"
    )
    
    return result

import json

def debate_pattern(question, num_agents=3):
    """Múltiplos agents respondem, depois debatem até consenso."""
    
    # Round 1: respostas independentes
    responses = []
    for i in range(num_agents):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": f"Você é o analista {i + 1}. Responda independentemente."},
                {"role": "user", "content": question},
            ],
            temperature=0.7,
        )
        responses.append(response.choices[0].message.content)
    
    # Round 2: cada agent vê as respostas dos outros e revisa
    revised = []
    for i, original in enumerate(responses):
        others = [reply for j, reply in enumerate(responses) if j != i]
        revision = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": "Revise sua resposta considerando as análises dos colegas. "
                    "Se concordar, mantenha. Se discordar, argumente por quê.",
                },
                {
                    "role": "user",
                    "content": f"Sua resposta: {original}\nRespostas dos colegas: {json.dumps(others, ensure_ascii=False)}",
                },
            ],
        )
        revised.append(revision.choices[0].message.content)
    
    # Round 3: sintetizar consenso
    final = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Sintetize o consenso dos analistas. Destaque onde concordam e onde divergem."},
            {"role": "user", "content": f"Análises finais:\n{json.dumps(revised, ensure_ascii=False)}"},
        ],
    )
    
    return final.choices[0].message.content

class SupervisoryArchitecture:
    def __init__(self):
        self.worker = WorkerAgent()
        self.supervisor = SupervisorAgent()
        self.max_interventions = 3
    
    def run(self, task):
        interventions = 0
        
        while interventions < self.max_interventions:
            # Worker tenta resolver
            result = self.worker.attempt(task)
            
            # Supervisor avalia
            assessment = self.supervisor.assess(task, result)
            
            if assessment["approved"]:
                return result
            elif assessment["action"] == "retry_with_guidance":
                # Supervisor dá orientação pro worker
                task = f"{task}\n\nGuidance: {assessment['guidance']}"
                interventions += 1
            elif assessment["action"] == "escalate":
                return self.escalate_to_human(task, result, assessment)
            elif assessment["action"] == "takeover":
                # Supervisor resolve ele mesmo
                return self.supervisor.solve(task)
        
        return self.escalate_to_human(task, result, "max_interventions_reached")

from dataclasses import dataclass
from datetime import datetime

@dataclass
class AgentMessage:
    sender: str
    receiver: str
    type: str  # "request", "response", "notification"
    content: dict
    correlation_id: str
    timestamp: datetime

# Exemplo de troca
msg1 = AgentMessage(
    sender="orchestrator",
    receiver="db_agent",
    type="request",
    content={"task": "Verificar replicação do PostgreSQL em db-prod-01"},
    correlation_id="task-2024-001",
    timestamp=datetime.now(),
)

import asyncio
from datetime import datetime

class Blackboard:
    def __init__(self):
        self.facts = {}  # {"server_cpu": 87.5, "diagnosis": "memory_leak"}
        self.lock = asyncio.Lock()
    
    async def write(self, agent_id, key, value, confidence=1.0):
        async with self.lock:
            self.facts[key] = {
                "value": value,
                "written_by": agent_id,
                "confidence": confidence,
                "timestamp": datetime.now(),
            }
    
    async def read(self, key):
        async with self.lock:
            return self.facts.get(key)
    
    async def get_all_facts(self):
        async with self.lock:
            return dict(self.facts)

from datetime import datetime, timedelta
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half-open"

class AgentCircuitBreaker:
    def __init__(self, failure_threshold=3, reset_timeout=60):
        self.failures = 0
        self.threshold = failure_threshold
        self.state = CircuitState.CLOSED
        self.opened_at = None
        self.reset_timeout = timedelta(seconds=reset_timeout)
    
    def call(self, agent, task):
        if self.state == CircuitState.OPEN:
            if self.opened_at and datetime.now() - self.opened_at >= self.reset_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                return {"error": "Agent circuit breaker open", "fallback": True}
        
        try:
            result = agent.run(task)
        except Exception:
            self.failures += 1
            self.opened_at = datetime.now()
            if self.state == CircuitState.HALF_OPEN or self.failures >= self.threshold:
                self.state = CircuitState.OPEN
            raise
        else:
            self.failures = 0
            self.opened_at = None
            self.state = CircuitState.CLOSED
            return result