Início A Filosofia Padrões
Documentação Oficial

A Filosofia da bee.arq

Entenda os princípios de coreografia de eventos e desacoplamento que tornam esta arquitetura modular, escalável e economicamente eficiente.

A bee.arq é um padrão arquitetural para a construção de sistemas de Inteligência Artificial complexos, inspirado na eficiência, especialização e comunicação de uma colmeia de abelhas. O foco é criar sistemas modulares, escaláveis, resilientes e economicamente eficientes, seguindo os princípios da **coreografia de eventos** e do **desacoplamento total dos componentes**.

A seguir, detalhamos cada conceito análogo e sua aplicação técnica.

A Floresta: O Ecossistema de Recursos

Analogia

Uma colmeia não existe no vácuo. Ela prospera numa floresta rica em recursos: flores (dados), água (computação) e um clima favorável (infraestrutura). A escolha da floresta é uma decisão estratégica que determina o potencial de crescimento e a resiliência da colmeia. Uma floresta pobre limita a colmeia, enquanto uma rica permite que ela floresça.

Termo Técnico

Plataforma de Nuvem e Seus Serviços Geridos (Cloud Provider). Embora a arquitetura `bee.arq` possa, em teoria, ser implementada num servidor monolítico ou num data center on-premise, a sua verdadeira essência de escalabilidade, resiliência e custo-eficiência é potencializada ao máximo num ambiente de nuvem. A "Floresta" é a plataforma de nuvem escolhida (AWS, Google Cloud, Azure, etc.).

Na Prática

A decisão de "instalar a colmeia" numa determinada "floresta" implica utilizar os serviços geridos que melhor se mapeiam para os componentes da `bee.arq`. A arquitetura foi concebida para ser nativa da nuvem (cloud-native).

A escolha da "Floresta" não prende a arquitetura, mas uma boa escolha acelera drasticamente o desenvolvimento e a evolução do "enxame", pois permite que a equipe se foque na lógica de negócio das abelhas, em vez de gerir a infraestrutura subjacente.

graph LR subgraph Cloud["☁️ Provedores de Nuvem"] Lambda("Lambda\n(AWS)") GCF("GCF\n(Google)") AzF("AzF\n(Azure)") SQS("SQS/SNS\n(AWS)") PubSub("Pub/Sub\n(Google)") ASB("ASB\n(Azure)") S3("S3\n(AWS)") GCS("GCS\n(Google)") Blob("Blob\n(Azure)") SageMaker("SageMaker\n(AWS)") VertexAI("Vertex AI\n(Google)") AzureML("Azure ML\n(Azure)") end subgraph BeeArq["🐝 Componentes bee.arq"] Operarias("Operárias") Feromonios("Feromônios") Polen("Pólen") Zangoes("Zangões") end %% Interligações compactas Operarias --> Lambda Operarias --> GCF Operarias --> AzF Feromonios -->|Eventos em| SQS Feromonios --> PubSub Feromonios --> ASB Polen -->|Armazenado em| S3 Polen --> GCS Polen --> Blob Zangoes -->|Hospedados em| Lambda Zangoes --> GCF Zangoes --> AzF Zangoes -->|Treinados com| SageMaker Zangoes --> VertexAI Zangoes --> AzureML

As Flores: A Origem do Pólen

Analogia

As flores são a fonte de vida para a colmeia. Elas não são parte da colmeia, mas existem na "Floresta" e oferecem o néctar e o pólen que as operárias coletam. Uma flor pode ser uma única planta ou um campo inteiro; pode ser perene ou florescer apenas sob certas condições. Quando uma flor "nasce" (gera dados), ela se torna um alvo para uma Operária.

Termo Técnico

Fonte de Dados Externa e Gatilho Inicial (Event Source). As "Flores" são os sistemas, aplicações ou ações humanas que geram os dados brutos (o "Pólen") que a colmeia irá processar. Elas são a origem de todo o fluxo de trabalho.

As Flores são dinâmicas e podem assumir qualquer forma:

Na Prática

A `bee.arq` é desenhada para reagir ao "nascimento" das flores. Uma Operária Inicial é configurada para "escutar" estas fontes de dados. Este é o ponto de entrada da arquitetura. A comunicação entre a Flor e a Operária é o primeiro e mais fundamental "Feromônio" do sistema.

graph LR subgraph Floresta [Ecossistema Externo] F1["Flor 1:
Upload em Bucket S3"] F2["Flor 2:
Webhook de CRM"] F3["Flor 3:
Trigger de Base de Dados"] end subgraph Colmeia O1(Operária Inicial) end F1 -->|Gatilho de Evento| O1 F2 -->|Gatilho de Evento| O1 F3 -->|Gatilho de Evento| O1 style Floresta fill:#d4edda,stroke:#155724,stroke-width:2px,stroke-dasharray: 5 5 style Colmeia fill:#fffbe6,stroke:#ffc107,stroke-width:2px

O Pólen: A Matéria-Prima Universal

Analogia

Na natureza, o pólen é a matéria-prima essencial, mas chega à colmeia em diversas formas, vindo de flores diferentes. As abelhas não pedem para a flor "pré-formatar" o pólen. Elas o coletam como está e o processo de transformação em **Mel** começa dentro da colmeia. O que não é útil é simplesmente ignorado.

Termo Técnico

Entidade de Dado Bruto e Agregado. O Pólen na `bee.arq` representa a unidade de dado no seu estado mais puro, qualquer que seja a sua origem ou formato: um ficheiro num bucket (multimédia, texto, binário), uma linha numa base de dados, um stream de eventos ou uma chamada de API. Ele é a representação do "problema" a ser resolvido, sem a necessidade de pré-classificação pelo sistema cliente.

Na Prática

A principal força do conceito de Pólen é a transferência de responsabilidade. O sistema cliente não precisa de saber o que fazer com os dados; ele apenas os entrega. Imaginemos que um departamento de contabilidade envia uma pasta com 100 ficheiros mistos (planilhas, PDFs de faturas, imagens de recibos).

  1. Entrega do Pólen: O sistema cliente apenas aponta para esta pasta.
  2. Identificação pela Operária Inicial: A primeira Operária do pipeline é acionada. A sua função é iterar sobre cada ficheiro (cada "grão de pólen"), identificar o seu tipo (ex: `.xlsx`, `.pdf`) e emitir um "Feromônio" específico para cada um.
  3. Descarte ou Processamento: Se um ficheiro for de um tipo desconhecido ou irrelevante para o contexto da Colmeia (ex: um ficheiro `.mp3` numa colmeia financeira), ele é ignorado ou movido para uma área de exceção. A colmeia decide o que é útil para si.
  4. Identificação e Classificação: Um ou mais Zangões podem ser acionados para identificar, modificar ou classificar o "Pólen" em sua origem.
  5. Evolução do Payload: O "Pólen" que conhecemos como o objeto de dados (o JSON ou interface tipada) não é o dado inicial, mas sim o **invólucro que é construído e enriquecido** em torno do dado bruto ao longo do pipeline.

O código abaixo mostra a evolução do payload do Pólen. Ele começa com informações mínimas e é enriquecido a cada passo pelas Operárias e Zangões.


# 1. Pólen Inicial (gerado pela primeira Operária)
pollen_inicial = {
    "pollen_id": "uuid-123",
    "source_uri": "s3://contabilidade/faturas/janeiro/fatura_xyz.pdf",
    "status": "RECEIVED"
}

# 2. Pólen após Zangão-OCR e Heurísticas (enriquecido pela Operária)
pollen_enriquecido = {
    "pollen_id": "uuid-123",
    "source_uri": "s3://contabilidade/faturas/janeiro/fatura_xyz.pdf",
    "status": "CLASSIFIED",
    "raw_text": "Texto extraído do PDF...",
    "doc_type": "invoice",
    "confidence": 0.98,
    "extracted_data": {
        "valor": 1950.50,
        "vencimento": "2025-06-30"
    }
}

# O modelo Pydantic representa o estado final e completo do Pólen
from pydantic import BaseModel, Field
from typing import List, Optional

class FinalPollen(BaseModel):
    pollen_id: str
    source_uri: str
    status: str
    raw_text: Optional[str] = None
    doc_type: Optional[str] = None
    extracted_data: Optional[dict] = None
    confidence_scores: List[float] = Field(default_factory=list)


// 1. Pólen Inicial (gerado pela primeira Operária)
interface InitialPollen {
  pollen_id: string;
  source_uri: string; // ex: 's3://bucket/key' ou 'db://table/id'
  status: 'RECEIVED';
}

// 2. Pólen após enriquecimento
interface EnrichedPollen extends InitialPollen {
  status: 'PROCESSED' | 'FAILED';
  raw_text?: string;
  doc_type?: string;
  extracted_data?: { [key: string]: any };
  confidence_scores?: number[];
  error_message?: string;
}

// O tipo final pode ser uma união ou uma interface completa
// para garantir a consistência no final do pipeline.
type Pollen = InitialPollen | EnrichedPollen;

A Colmeia: O Contexto Delimitado do Sistema

Analogia

Na natureza, a colmeia é a estrutura que define o lar e o campo de atuação de uma sociedade de abelhas. Ela não dá ordens, mas representa o universo onde as regras de interação e os objetivos (como a produção de mel) existem. A vida da colmeia emerge da ação coordenada das suas integrantes.

Termo Técnico

Contexto de Execução e Limite Transacional. A Colmeia não é um componente ativo, mas sim a representação abstrata de um sistema ou de um pipeline de negócio completo. Ela delimita quais Operárias, Zangões e qual Rainha atuam juntos para resolver um problema específico. O pipeline "dentro" de uma Colmeia ganha vida quando uma Operária Inicial é acionada por um evento externo. No contexto da `bee.arq`, uma operária inicial é qualquer operária que polinize ou receba feromônios na colmeia; na prática, qualquer função que receba dados de entrada, mensagem ou evento que inicie um ou vários pipelines na colmeia.

Na Prática

Pense na Colmeia como um projeto no seu repositório de código ou um "Bounded Context" do Domain-Driven Design (DDD). Ela agrupa todos os recursos de nuvem (funções de eventos, filas de mensagens, APIs, Banco de Dados e I.A.s menores) que servem a um propósito. A política de tratamento de erros e a lógica de recuperação não são uma função da Colmeia, mas sim uma propriedade inerente à infraestrutura que a suporta (ex.: as políticas de retry de um serviço de fluxo de trabalho como AWS Step Functions, Google Cloud Workflows ou Azure Logic Apps, ou de uma fila de mensagens mortas).

A Colmeia tem crescimento orgânico vertical e horizontal. No entanto, a natureza da arquitetura `bee.arq` favorece preferencialmente o crescimento horizontal, que estabelece um menor custo com maior aproveitamento computacional.

graph LR subgraph Colmeia [Colmeia - Contexto do Problema] O1(Operária Inicial) -->|Chama| Z1(Zangão A) Z1 -->|Retorna| O1 O1 -->|Chama| O2(Operária de Enriquecimento) O2 -->|Chama| Z2(Zangão B) Z2 -->|Retorna| O2 O2 -->|Chama| R(Rainha) end Evento[Evento Externo
ex: Upload de Arquivo] --> O1 style Colmeia fill:#fffbe6,stroke:#ffc107,stroke-width:2px,stroke-dasharray: 5 5 style Evento fill:#e3f2fd,stroke:#90caf9

As Operárias: O Coração da Coreografia

Analogia

As operárias são as trabalhadoras incansáveis e o sistema nervoso da colmeia. Elas coletam o **Pólen** (dados brutos) e iniciam o processo de transformação em **Mel**. Elas executam tarefas determinísticas (limpar, construir, organizar) e se comunicam para criar um fluxo de trabalho inteligente. Elas são a ligação entre a matéria-prima e a decisão final da Rainha, conduzindo todo o processo de produção.

Termo Técnico

Unidade de Processamento Determinístico e Roteamento. A Operária é o componente mais crucial da `bee.arq`. Sem elas, os Zangões são especialistas isolados sem trabalho, e a Rainha nunca recebe o "Mel" preparado para tomar uma decisão. Elas são responsáveis por todo o fluxo de trabalho, executando lógica de negócio determinística.

As principais características de uma Operária são:

Na Prática

O poder das Operárias está na sua composição para criar fluxos complexos, transformando o "Pólen" em "Mel" passo a passo. O diagrama abaixo ilustra os diferentes papéis que as operárias podem desempenhar.

Considere o seguinte cenário:

  1. Operária de Ingestão (Inicial): É acionada pelo upload de um vídeo. A sua tarefa é extrair a trilha de áudio, emitindo um evento de "Áudio Extraído". Esta é a primeira etapa na transformação do Pólen.
  2. Operária de Transcrição (Enriquecimento): Recebe o áudio, chama um `ZangaoDeTranscricao` (que adiciona inteligência), recebe o texto de volta e o adiciona ao "Pólen", continuando o seu refinamento.
  3. Operária de Validação (Enriquecimento): A mesma Operária, após receber o texto, pode executar uma lógica determinística para validar o conteúdo, garantindo a qualidade do "Pólen" que está a ser processado.
  4. Operária de Consolidação (Final): No final da cadeia, esta Operária recebe o "Pólen" já muito refinado e a sua função é consolidá-lo na forma final de **"Mel Puro"**: um prompt factual, coeso e pronto para ser enviado à `Rainha`.
flowchart TD subgraph sg_inicial["Papel: Operária Inicial"] OD1["Processa Dados Iniciais"] OI1("Recebe Evento Externo") PZ1{"Precisa de Zangão?"} Z1("Zangão Especialista") PR1("Resultado Consolidado") RO1("Chama Próxima Operária") end subgraph sg_enriquecimento["Papel: Operária de Enriquecimento"] ODE1["Enriquece / Transforma"] OE1("Recebe Pólen") PZE1{"Precisa de Zangão?"} Z2("Zangão Especialista") PRE1("Resultado Consolidado") ROE1("Chama Próxima Operária") end subgraph sg_consolidacao["Papel: Operária de Consolidação"] OC1("Recebe Pólen Final") ODC1{"Consolida e Formata"} R("Chama a Rainha") end EventoExterno["Evento Externo"] --> OI1 OI1 --> OD1 OD1 --> PZ1 PZ1 -- Sim --> Z1 Z1 --> PR1 PZ1 -- Não --> PR1 PR1 --> RO1 RO1 --> OE1 OE1 --> ODE1 ODE1 --> PZE1 PZE1 -- Sim --> Z2 Z2 --> PRE1 PZE1 -- Não --> PRE1 PRE1 --> ROE1 ROE1 --> OC1 OC1 --> ODC1 ODC1 --> R RO1 --> OC1 style sg_consolidacao fill:#C8E6C9 style sg_inicial fill:#fff3cd,stroke:#ffc107,stroke-width:2px style sg_enriquecimento fill:#cfe2ff,stroke:#0d6efd,stroke-width:2px

O exemplo de código abaixo representa uma Operária que combina múltiplas responsabilidades: chamar Zangões, aplicar lógica determinística e produzir "Geleia Real" para observabilidade.


# Exemplo de uma Operária em Python com produção de Geleia Real
import requests
import logging
import json

# Configura um logger para outputar JSON (Geleia Real)
logger = logging.getLogger("GeleiaRealLogger")
logger.setLevel(logging.INFO)

class DocumentPipelineOperaria:
    def __init__(self, pollen):
        self.pollen = pollen
        self.pollen_id = pollen.get("pollen_id", "unknown")

    def _log_geleia(self, status, actor, details, data=None):
        log_entry = { "pollen_id": self.pollen_id, "status": status, "actor": actor, "details": details, "pollen_snapshot": data or self.pollen }
        logger.info(json.dumps(log_entry))

    def _call_zangao(self, url, payload):
        actor_name = url.split("/")[-1]
        try:
            self._log_geleia("CALLING_ZANGAO", actor_name, f"Enviando dados para {url}")
            response = requests.post(url, json=payload, timeout=15)
            response.raise_for_status()
            result = response.json()
            self._log_geleia("ZANGAO_SUCCESS", actor_name, "Resposta recebida com sucesso", data=result)
            return result
        except requests.RequestException as e:
            self._log_geleia("ZANGAO_FAILED", actor_name, str(e))
            raise

    def run_pipeline(self):
        try:
            self._log_geleia("PIPELINE_STARTED", self.__class__.__name__, "Iniciando pipeline de documento")
            
            # 1. Chama o Zangão-OCR
            ocr_result = self._call_zangao("http://zangao-ocr.internal/extract-text", self.pollen)
            self.pollen['raw_text'] = ocr_result['text']

            # 2. Realiza tarefa determinística (heurística)
            if "fatura" in self.pollen['raw_text'].lower():
                self.pollen['heuristic_tag'] = "invoice"
            self._log_geleia("HEURISTIC_APPLIED", self.__class__.__name__, "Tag heurística adicionada")
            
            # 3. Chama o Zangão-Classifier
            classifier_result = self._call_zangao("http://zangao-classifier.internal/classify", self.pollen)
            self.pollen['doc_type'] = classifier_result['type']

            self._log_geleia("PIPELINE_SUCCESS", self.__class__.__name__, "Pipeline concluído")
            return self.pollen
        except Exception as e:
            self._log_geleia("PIPELINE_FAILED", self.__class__.__name__, f"Erro geral no pipeline: {e}")
            raise


// Exemplo de uma Operária em TypeScript com produção de Geleia Real
type Pollen = { [key: string]: any };

// Logger simples (poderia ser Winston, Pino, etc.)
const logger = {
    info: (logEntry: object) => console.log(JSON.stringify(logEntry, null, 2)),
    error: (logEntry: object) => console.error(JSON.stringify(logEntry, null, 2)),
};

class DocumentPipelineOperaria {
    private pollen: Pollen;
    private pollenId: string;

    constructor(pollen: Pollen) {
        this.pollen = pollen;
        this.pollenId = pollen.pollen_id || "unknown";
    }

    private logGeleia(status: string, actor: string, details: string, data?: any) {
        const logEntry = { pollen_id: this.pollenId, timestamp: new Date().toISOString(), status, actor, details, pollen_snapshot: data || this.pollen };
        if (status.includes("FAILED")) logger.error(logEntry);
        else logger.info(logEntry);
    }

    private async callZangao(url: string, payload: any): Promise<any> {
        const actorName = url.split("/").pop() || "unknown-zangao";
        try {
            this.logGeleia("CALLING_ZANGAO", actorName, `Enviando dados para ${url}`);
            const response = await fetch(url, {
                method: 'POST',
                body: JSON.stringify(payload),
                headers: { 'Content-Type': 'application/json' },
            });
            if (!response.ok) throw new Error(response.statusText);
            const result = await response.json();
            this.logGeleia("ZANGAO_SUCCESS", actorName, "Resposta recebida com sucesso", result);
            return result;
        } catch (error: any) {
            this.logGeleia("ZANGAO_FAILED", actorName, error.message);
            throw error;
        }
    }

    public async runPipeline(): Promise<Pollen> {
        const actorName = this.constructor.name;
        try {
            this.logGeleia("PIPELINE_STARTED", actorName, "Iniciando pipeline de documento");

            const ocrResult = await this.callZangao("http://zangao-ocr.internal/extract-text", this.pollen);
            this.pollen.raw_text = ocrResult.text;

            if (this.pollen.raw_text.toLowerCase().includes("fatura")) {
                this.pollen.heuristic_tag = "invoice";
            }
            this.logGeleia("HEURISTIC_APPLIED", actorName, "Tag heurística adicionada");

            const classifierResult = await this.callZangao("http://zangao-classifier.internal/classify", this.pollen);
            this.pollen.doc_type = classifierResult.type;

            this.logGeleia("PIPELINE_SUCCESS", actorName, "Pipeline concluído");
            return this.pollen;
        } catch (error: any) {
            this.logGeleia("PIPELINE_FAILED", actorName, `Erro geral no pipeline: ${error.message}`);
            throw error;
        }
    }
}

Os Zangões: Especialistas Puros de Inteligência

Analogia

O papel do zangão na colmeia é singular e vital. Ele não se envolve nas tarefas de construção ou limpeza, mas a sua função é crítica para o futuro da colmeia. Na `bee.arq`, o Zangão é a ferramenta especialista convocada por uma Operária para aplicar uma capacidade cognitiva superior ao **Pólen**, refinando-o e dando um passo essencial na produção do **Mel**.

Termo Técnico

Serviço de I.A. Especialista, Autônomo e Reutilizável. Um Zangão é um serviço que encapsula uma capacidade cognitiva. A regra de ouro mantém-se: ele recebe o "Pólen" de uma Operária e sempre devolve o "Pólen" enriquecido para a mesma Operária que o chamou. Este princípio garante o seu total desacoplamento e o torna 100% reutilizável e interoperável entre diferentes Colmeias e fluxos.

A função primordial de um Zangão é aprimorar, detetar, classificar ou raciocinar sobre o "Pólen" para que a decisão do próximo passo seja a mais assertiva possível. No entanto, a sua forma e escala são extremamente flexíveis:

Na Prática

A beleza dos Zangões está na sua diversidade e especialização. Qualquer um dos exemplos abaixo pode ser um Zangão na arquitetura:

  1. Zangão de Visão Computacional: Uma Operária envia uma imagem de um produto. O Zangão, contendo uma CNN, retorna as coordenadas de um defeito encontrado na imagem.
  2. Zangão de Remasterização de Áudio: Uma Operária envia um arquivo de áudio com ruído. O Zangão aplica uma série de filtros de frequência e algoritmos de IA para limpar o áudio e retorna a versão remasterizada.
  3. Zangão de Análise Semântica: Uma Operária envia o texto de um contrato. O Zangão, que contém um modelo de NLP treinado, não apenas classifica o documento, mas extrai e retorna um JSON estruturado com as partes de "cláusulas de rescisão" e "multas aplicáveis".
  4. Zangão-RAG: Uma Operária envia uma pergunta complexa de um cliente. Este Zangão, que é um serviço completo com o seu próprio banco de dados vetorial e LLM, busca a informação na sua base de conhecimento, raciocina sobre ela e retorna uma resposta final e elaborada para a Operária.

O exemplo de código abaixo ilustra a implementação de um Zangão especialista em transcrição de áudio, mas é crucial entender que a sua complexidade interna pode ser muito maior.


# zangao_transcricao.py (usando FastAPI e Hugging Face Transformers)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from transformers import pipeline
import torch
import librosa # Biblioteca para carregar áudio

# --- Inicialização do Modelo (carrega apenas uma vez) ---
# Define o dispositivo (GPU se disponível, senão CPU)
device = "cuda:0" if torch.cuda.is_available() else "cpu"

# Carrega o pipeline de transcrição com o modelo especificado
# Nota: Este download pode demorar na primeira execução.
transcription_pipeline = pipeline(
    "automatic-speech-recognition",
    model="distil-whisper/distil-large-v3",
    torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
    device=device,
)
# ---------------------------------------------------------

app = FastAPI(
    title="Zangão de Transcrição de Áudio",
    description="Um serviço especialista que transcreve áudio usando distil-whisper."
)

# --- Contratos de Dados (Pólen de Entrada/Saída) ---
class AudioInput(BaseModel):
    # A Operária informa a localização do arquivo de áudio.
    audio_path: str 
    # Ex: "s3://bucket/audio.mp3", "gs://bucket/audio.mp3"

class TranscriptionOutput(BaseModel):
    transcribed_text: str
    language: str

# --- Endpoint do Zangão ---
@app.post("/transcribe", response_model=TranscriptionOutput)
def transcribe_audio(payload: AudioInput):
    """
    Recebe o caminho de um arquivo de áudio, carrega-o e
    retorna a transcrição.
    """
    try:
        # Em um cenário real, aqui entraria a lógica para baixar o
        # ficheiro de um serviço de armazenamento na nuvem.
        # Para este exemplo, vamos assumir que o caminho é local.
        local_audio_path = payload.audio_path
        
        # Carrega o áudio e garante a taxa de amostragem correta (16kHz para o Whisper)
        audio_input, sample_rate = librosa.load(local_audio_path, sr=16000)

        # O Zangão executa sua única função: transcrever.
        result = transcription_pipeline(audio_input)
        
        # Retorna o resultado para a Operária que o chamou.
        return {
            "transcribed_text": result["text"],
            "language": "detectado" # O Whisper detecta o idioma
        }
    except FileNotFoundError:
        raise HTTPException(status_code=404, detail=f"Arquivo de áudio não encontrado em: {payload.audio_path}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Erro durante a transcrição: {str(e)}")


// zangao_cliente_transcricao.ts (usando Express e Axios)
// Este exemplo mostra um Zangão em TS atuando como um cliente para o serviço Python de ML.
// Este é um padrão comum para integrar ecossistemas diferentes.

import express, { Request, Response } from 'express';
import axios from 'axios';

const app = express();
app.use(express.json());

// O endereço do nosso Zangão especialista em Python
const PYTHON_TRANSCRIPTION_SERVICE_URL = "http://localhost:8000/transcribe";

// --- Contratos de Dados (Interface do Pólen) ---
interface AudioInput {
    audio_path: string;
}

interface TranscriptionOutput {
    transcribed_text: string;
    language: string;
}

// --- Endpoint do Zangão (que delega para o especialista) ---
app.post('/transcribe', async (req: Request, res: Response) => {
    try {
        const payload: AudioInput = req.body;

        console.log(`Recebida solicitação para transcrever: ${payload.audio_path}`);
        console.log(`Encaminhando para o Zangão especialista em Python...`);

        // A função deste Zangão é puramente chamar o outro serviço.
        // Ele age como um proxy ou um adaptador na Colmeia.
        const response = await axios.post<TranscriptionOutput>(
            PYTHON_TRANSCRIPTION_SERVICE_URL,
            payload
        );

        // Retorna a resposta do especialista para a Operária que o chamou.
        res.json(response.data);

    } catch (error: any) {
        // Se o serviço Python falhar, este Zangão repassa o erro.
        console.error("Erro ao comunicar com o serviço de transcrição:", error.message);
        const status = error.response?.status || 500;
        const detail = error.response?.data?.detail || "Erro interno no serviço de transcrição.";
        res.status(status).json({ detail });
    }
});

const PORT = 3001;
app.listen(PORT, () => {
    console.log(`Zangão Cliente de Transcrição (TS) ouvindo na porta ${PORT}`);
    console.log(`Redirecionando chamadas para: ${PYTHON_TRANSCRIPTION_SERVICE_URL}`);
});

Os Feromônios: O Sistema de Comunicação

Analogia

Os feromônios são os sinais químicos que governam a colmeia. Uma abelha libera um feromônio não para dar uma ordem direta, mas para transmitir uma informação ("fonte de néctar encontrada aqui", "perigo iminente"). Outras abelhas reagem a esse sinal de forma autônoma, criando um comportamento coletivo complexo e descentralizado.

Termo Técnico

Protocolo de Transmissão e Eventos. Na `bee.arq`, um "Feromônio" é qualquer mecanismo de comunicação que transfere o "Pólen" (dados) e dispara a ação do próximo agente. A arquitetura é flexível quanto ao protocolo, permitindo o uso da ferramenta certa para cada tarefa.

Na Prática

A escolha do "Feromônio" é uma decisão de design baseada nos requisitos do pipeline. Para tarefas em tempo real, uma Operária pode fazer uma chamada HTTP direta a um Zangão. Para o processamento de grandes volumes de dados, a mesma Operária pode, em vez disso, publicar uma mensagem numa fila (como AWS SQS, Google Pub/Sub ou Azure Service Bus). O importante é que o "Pólen" (o contrato de dados) permaneça consistente, independentemente do meio de transporte.


# operaria_que_chama.py (usando SQS para comunicação assíncrona)
import boto3
import json

sqs_client = boto3.client('sqs')
QUEUE_URL = 'https://sqs.us-east-1.amazonaws.com/123456789012/proxima-operaria-queue'

def enviar_feromonio_assincrono(pollen):
    """
    Publica o pólen enriquecido como uma mensagem (feromônio) para a próxima
    etapa do pipeline de forma assíncrona.
    """
    try:
        sqs_client.send_message(
            QueueUrl=QUEUE_URL,
            MessageBody=json.dumps(pollen)
        )
        print("Feromônio assíncrono enviado com sucesso!")
    except Exception as e:
        print(f"Erro ao enviar feromônio: {e}")
        raise

# operaria_que_recebe.py (em outro serviço/lambda, acionada pela fila)
def lambda_handler_assincrono(event, context):
    for record in event['Records']:
        pollen = json.loads(record['body'])
        # Processa o pólen recebido da fila
        print(f"Feromônio recebido! Processando file_id: {pollen.get('file_id')}")
        # ... lógica da próxima operária ...


// operaria_que_chama.ts (usando SQS para comunicação assíncrona)
import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";

const sqsClient = new SQSClient({});
const QUEUE_URL = 'https://sqs.us-east-1.amazonaws.com/123456789012/proxima-operaria-queue';

async function enviarFeromonioAssincrono(pollen: object): Promise<void> {
    const command = new SendMessageCommand({
        QueueUrl: QUEUE_URL,
        MessageBody: JSON.stringify(pollen),
    });
    try {
        await sqsClient.send(command);
        console.log("Feromônio assíncrono enviado com sucesso!");
    } catch (error) {
        console.error("Erro ao enviar feromônio:", error);
        throw error;
    }
}

// operaria_que_recebe.ts (em outro serviço/lambda, acionada pela fila)
import { SQSEvent } from "aws-lambda";

export const handler = async (event: SQSEvent): Promise<void> => {
    for (const record of event.Records) {
        const pollen = JSON.parse(record.body);
        // Processa o pólen recebido da fila
        console.log(`Feromônio recebido! Processando file_id: ${pollen.file_id}`);
        // ... lógica da próxima operária ...
    }
};

A Rainha: O Propósito da Colmeia

Analogia

Na natureza, toda a atividade da colmeia gira em torno de um único propósito: garantir a sobrevivência e a prosperidade da Rainha e, por extensão, de toda a sociedade. Ela não consome o **Pólen** bruto, mas sim o **Mel Puro**, o produto final e altamente nutritivo, preparado pelas operárias. Com base neste alimento perfeito, ela executa a sua função de maior impacto.

Termo Técnico

Provedor de Contexto Factual e Função de Síntese Resiliente. Na `bee.arq`, a Rainha é o propósito do pipeline. Todo o trabalho de transformação do "Pólen" em "Mel" realizado pelas Operárias e Zangões visa entregar um produto final, factual e estruturado, para que a Rainha possa tomar a decisão mais assertiva e de maior valor de negócio.

Isto representa uma mudança fundamental na forma como interagimos com as LLMs:

Na Prática

A implementação da Rainha foca-se em criar um "prompt factual" dinâmico a partir do "Mel" e em gerir a comunicação com múltiplos provedores de LLM. O código de exemplo ilustra a lógica de fallback, que é o cerne da sua resiliência.


# Exemplo de uma Rainha em Python com fallback
import requests
import os
import json

# Configuração das LLMs (poderia vir de um serviço de segredos)
LLM_PROVIDERS = [
    {
        "name": "OpenAI",
        "url": "https://api.openai.com/v1/chat/completions",
        "headers": {"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}"},
        "model": "gpt-4"
    },
    {
        "name": "Google",
        "url": f"https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:generateContent?key={os.getenv('GEMINI_API_KEY')}",
        "headers": {"Content-Type": "application/json"}
    }
]

class Queen:
    def __init__(self, system_prompt):
        self.system_prompt = system_prompt

    def make_decision(self, honey_data): # Agora recebe "Mel"
        errors = {}
        for provider in LLM_PROVIDERS:
            try:
                # O prompt é construído com dados factuais e enriquecidos
                body = self._adapt_body(provider, honey_data)
                
                response = requests.post(
                    provider['url'], 
                    json=body, 
                    headers=provider['headers'],
                    timeout=30
                )
                response.raise_for_status()
                
                return self._extract_response(provider['name'], response.json())

            except requests.exceptions.RequestException as e:
                print(f"Erro ao chamar {provider['name']}: {e}")
                errors[provider['name']] = str(e)
        
        raise Exception(f"Falha em todos os provedores de LLM: {errors}")

    def _adapt_body(self, provider, honey_data):
        content = f"{self.system_prompt}\n\nCONTEXTO FACTUAL (MEL PURO):\n{json.dumps(honey_data, indent=2)}"
        if provider['name'] == 'OpenAI':
            return {"model": provider['model'], "messages": [{"role": "user", "content": content}]}
        if provider['name'] == 'Google':
            return {"contents": [{"parts": [{"text": content}]}]}
        return {}

    def _extract_response(self, provider_name, data):
        if provider_name == 'OpenAI':
            return data['choices'][0]['message']['content']
        if provider_name == 'Google':
            return data['candidates'][0]['content']['parts'][0]['text']
        return ""


// Exemplo de uma Rainha em TypeScript com fallback
interface LLMProvider {
    name: string;
    url: string;
    headers: Record<string, string>;
    adaptBody: (prompt: string, honey: any) => Record<string, any>;
    extractResponse: (responseData: any) => string;
}

const LLM_PROVIDERS: LLMProvider[] = [
    {
        name: "OpenAI",
        url: "https://api.openai.com/v1/chat/completions",
        headers: { "Authorization": `Bearer ${process.env.OPENAI_API_KEY}` },
        adaptBody: (prompt, honey) => ({
            model: "gpt-4",
            messages: [{ role: "user", content: `${prompt}\n\nCONTEXTO FACTUAL (MEL PURO):\n ${JSON.stringify(honey, null, 2)}` }]
        }),
        extractResponse: (data) => data.choices[0].message.content,
    },
    // Adicione outros provedores...
];

export class Queen {
    private systemPrompt: string;

    constructor(systemPrompt: string) {
        this.systemPrompt = systemPrompt;
    }

    public async makeDecision(honey: any): Promise<string> {
        const errors: Record<string, string> = {};

        for (const provider of LLM_PROVIDERS) {
            try {
                const body = provider.adaptBody(this.systemPrompt, honey);
                const response = await fetch(provider.url, {
                    method: 'POST',
                    headers: { ...provider.headers, 'Content-Type': 'application/json' },
                    body: JSON.stringify(body),
                });

                if (!response.ok) {
                    throw new Error(`API error: ${response.statusText}`);
                }

                const responseData = await response.json();
                return provider.extractResponse(responseData);

            } catch (error: any) {
                console.error(`Erro ao chamar ${provider.name}:`, error.message);
                errors[provider.name] = error.message;
            }
        }
        throw new Error(`Falha em todos os provedores de LLM: ${JSON.stringify(errors)}`);
    }
}

A Geleia Real: Observabilidade e Aprendizado Contínuo

Analogia

A geleia real é o superalimento da colmeia, mas também é um indicador da sua saúde e produtividade. Ao analisar a geleia, um apicultor (o desenvolvedor) entende o que a colmeia está a processar, se está saudável ou se precisa de intervenção. Ela é a fonte de verdade sobre o estado interno da colmeia, usada tanto para criar novas rainhas (melhorar o sistema) quanto para diagnosticar problemas.

Termo Técnico

Mecanismo de Observabilidade, Rastreabilidade e Aprendizado Contínuo. A "Geleia Real" na `bee.arq` transcende o re-treinamento. É o dado primordial para todo o ciclo de vida do desenvolvimento e operações (DevOps/MLOps). Cada Operária, ao concluir uma etapa, deve produzir "geleia real".

Na Prática

A "Geleia Real" pode ser implementada de várias formas, dependendo da necessidade: logs estruturados em JSON enviados para um serviço de logging (Amazon CloudWatch, Google Cloud Logging, Azure Monitor) ou para uma stack de observabilidade como o ELK; registos numa tabela de um banco de dados NoSQL (Amazon DynamoDB, Google Firestore, Azure Cosmos DB) para auditoria; ou até mesmo ficheiros de texto num serviço de armazenamento de objetos (AWS S3, Google Cloud Storage, Azure Blob Storage) para análise de big data. O importante é que cada agente deixe o seu rasto.


# Exemplo de produção de Geleia Real (Log Estruturado) numa Operária
import logging
import json

# Configura um logger para outputar JSON
logger = logging.getLogger("GeleiaRealLogger")
logger.setLevel(logging.INFO)

class OperariaComGeleia:
    def __init__(self, pollen):
        self.pollen = pollen
        self.pollen_id = pollen.get("pollen_id", "unknown")

    def log_geleia_real(self, status, actor, details, data=None):
        log_entry = {
            "pollen_id": self.pollen_id,
            "status": status,
            "actor": actor,
            "details": details,
            "pollen_snapshot": data or self.pollen
        }
        # Em produção, um handler enviaria isso para o CloudWatch, S3, etc.
        logger.info(json.dumps(log_entry))

    def run(self):
        try:
            self.log_geleia_real("STARTED", "OperariaDeOCR", "Iniciando processo de OCR")
            # ... lógica para chamar o Zangão-OCR ...
            # self.pollen = ocr_operaria.run()
            self.log_geleia_real("SUCCESS", "OperariaDeOCR", "OCR concluído com sucesso")
            return self.pollen
        except Exception as e:
            self.log_geleia_real("FAILED", "OperariaDeOCR", str(e))
            raise

# processo_de_retreinamento.py (permanece o mesmo, consome a "geleia")
# ... (código de re-treinamento com a Hugging Face)


// Exemplo de produção de Geleia Real (Log Estruturado) numa Operária
import { randomUUID } from "crypto";

// Um logger simples que envia para o console, mas poderia ser Winston, Pino, etc.
// para enviar para um serviço de log.
class GeleiaRealLogger {
    log(logEntry: object) {
        console.log(JSON.stringify(logEntry, null, 2));
    }
}

const logger = new GeleiaRealLogger();

class OperariaComGeleia {
    private pollen: any;
    private pollenId: string;

    constructor(pollen: any) {
        this.pollen = pollen;
        this.pollenId = pollen.pollen_id || randomUUID();
    }

    private logGeleiaReal(status: 'STARTED' | 'SUCCESS' | 'FAILED', actor: string, details: string, data?: any) {
        const logEntry = {
            pollen_id: this.pollenId,
            timestamp: new Date().toISOString(),
            status,
            actor,
            details,
            pollen_snapshot: data || this.pollen,
        };
        logger.log(logEntry);
    }

    public async run(): Promise<any> {
        try {
            this.logGeleiaReal("STARTED", "OperariaDeOCR", "Iniciando processo de OCR");
            // ... lógica para chamar o Zangão-OCR ...
            // this.pollen = await ocrOperaria.run();
            self.logGeleiaReal("SUCCESS", "OperariaDeOCR", "OCR concluído com sucesso");
            return self.pollen;
        } catch (error: any) {
            self.logGeleiaReal("FAILED", "OperariaDeOCR", error.message);
            throw error;
        }
    }
}

// processo_de_retreinamento.ts (permanece o mesmo, consome a "geleia")
// ... (código de re-treinamento)