Implementação de Zangão

# contrato_zangao.py
from pydantic import BaseModel, Field
from typing import Optional, Dict, Any
from datetime import date, datetime

# 1. Definição do contrato do Zangão
class InvoicePollenInput(BaseModel):
    pollen_id: str = Field(..., description="ID único do pólen")
    text_content: str = Field(..., description="Conteúdo textual para análise")

class ExtractedData(BaseModel):
    valor_total: Optional[float] = Field(None, description="Valor total da fatura")
    data_vencimento: Optional[date] = Field(None, description="Data de vencimento")

class InvoicePollenOutput(BaseModel):
    pollen_id: str = Field(..., description="ID do pólen processado")
    extracted_data: ExtractedData = Field(..., description="Dados extraídos")
    model_version: str = Field("1.2.0", description="Versão do modelo usado")

# 2. Definição do Feromônio que transportará o resultado
class Feromonio(BaseModel):
    event_version: str = Field("1.0", description="Versão do evento")
    timestamp: datetime = Field(..., description="Momento do processamento")
    trace_id: str = Field(..., description="ID para rastreamento distribuído")
    source: str = Field(..., description="Origem do evento") 
    pollen_id: str = Field(..., description="ID do pólen relacionado")
    type: str = Field(..., description="Tipo do evento")
    payload: Dict[str, Any] = Field(..., description="Dados do evento")

# 3. Versão estendida do contrato (compatível com versão anterior)
class InvoicePollenInputV2(InvoicePollenInput):
    metadata: Optional[Dict[str, Any]] = Field(
        None,
        description="Metadados adicionais para processamento",
        example={"priority": "high", "origin": "scanned_document"}
    )
    
    class Config:
        schema_extra = {
            "description": "Contrato para processamento de faturas",
            "examples": [
                {
                    "pollen_id": "uuid-123",
                    "text_content": "Fatura XYZ...",
                    "metadata": {"priority": "high"}
                }
            ]
        }

# Como a Operária cria o Feromônio após receber a resposta do Zangão
# zangao_response = InvoicePollenOutput(...)
# feromonio_para_proxima_etapa = Feromonio(
#     ...,
#     payload=zangao_response.model_dump() 
# )

# operaria-lambda.py (A que chama)
import boto3
import json
from tenacity import retry, stop_after_attempt, wait_exponential

lambda_client = boto3.client('lambda')

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def chamar_zangao_diretamente(pollen):
    """Chama um Zangão com política de retry"""
    try:
        response = lambda_client.invoke(
            FunctionName='arn:aws:lambda:us-east-1:123:function:zangao-classifier',
            InvocationType='RequestResponse', # Invocação síncrona
            Payload=json.dumps(pollen)
        )
        response_payload = json.loads(response['Payload'].read())
        
        # Verifica se houve erro na execução
        if 'errorMessage' in response_payload:
            raise Exception(response_payload['errorMessage'])
            
        return response_payload
    except Exception as e:
        print(f"Falha ao chamar Zangão: {str(e)}")
        raise

# zangao-lambda.py (A que é chamada)
def handler(event, context):
    # 'event' é o payload (pollen) enviado pela Operária
    pollen = event 
    
    # Health check endpoint
    if pollen.get('action') == 'health_check':
        return {
            'status': 'healthy',
            'version': '1.2.0',
            'timestamp': datetime.utcnow().isoformat()
        }
    
    # ... executa a lógica do Zangão ...
    try:
        resultado = {"doc_type": "contract", "confidence": 0.99}
        return resultado
    except Exception as e:
        return {
            'errorMessage': str(e),
            'errorType': type(e).__name__,
            'stackTrace': traceback.format_exc().splitlines()
        }

# zangao_resnet.py (FastAPI com PyTorch e TorchVision)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import torch
import torchvision.transforms as T
from torchvision.models import resnet18, ResNet18_Weights
from PIL import Image
import requests
from io import BytesIO
import logging

# Configuração de logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 1. Carregar modelo pré-treinado (ou seu modelo proprietário)
weights = ResNet18_Weights.IMAGENET1K_V1
model = resnet18(weights=weights)
model.eval() # Colocar em modo de avaliação

# 2. Definir as transformações de imagem necessárias
preprocess = weights.transforms()

app = FastAPI(
    title="Zangão ResNet18",
    description="Serviço especializado em classificação de imagens",
    version="1.2.0"
)

class ImageInput(BaseModel):
    image_url: str
    metadata: Optional[dict] = None

@app.post("/classify-image")
def classify_image(data: ImageInput):
    """Endpoint principal para classificação de imagens"""
    try:
        logger.info(f"Processando imagem: {data.image_url}")
        
        # Em um caso real, esta URL poderia ser uma URI de S3/GCS
        response = requests.get(data.image_url, stream=True)
        response.raise_for_status()
        
        img = Image.open(BytesIO(response.content)).convert("RGB")
        
        # Aplicar pré-processamento
        batch = preprocess(img).unsqueeze(0)
        
        # Realizar predição
        with torch.no_grad():
            prediction = model(batch).squeeze(0).softmax(0)
            class_id = prediction.argmax().item()
            score = prediction[class_id].item()
            category_name = weights.meta["categories"][class_id]

        return {
            "predicted_category": category_name,
            "confidence": score,
            "model_version": "resnet18_v1",
            "metadata": data.metadata
        }
    except Exception as e:
        logger.error(f"Erro ao processar imagem: {str(e)}")
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
def health_check():
    """Endpoint para verificação de saúde"""
    return {
        "status": "healthy",
        "model": "resnet18",
        "version": "1.2.0"
    }

# Python usando tenacity
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type((TimeoutError, ConnectionError))
)
def call_zangao_with_circuit(pollen):
    """Chama um Zangão com política de retry inteligente"""
    # Implementação da chamada aqui
    pass

def process_with_fallback(pollen):
    """Tenta o Zangão principal, depois fallbacks"""
    strategies = [
        primary_zangao,
        fallback_zangao_1,
        fallback_zangao_2,
        ultimate_fallback
    ]
    
    last_error = None
    for strategy in strategies:
        try:
            return strategy(pollen)
        except Exception as e:
            print(f"Falha com estratégia {strategy.__name__}: {e}")
            last_error = e
    
    raise Exception("Todas estratégias falharam") from last_error

@app.get("/health")
def health_check():
    """Endpoint de health check"""
    try:
        # Verifica se o modelo está carregado
        if not model:
            raise Exception("Modelo não carregado")
            
        # Teste de predição simples
        test_input = torch.rand(1, 3, 224, 224)
        with torch.no_grad():
            model(test_input)
            
        return {
            "status": "healthy",
            "version": "1.2.0",
            "timestamp": datetime.utcnow().isoformat()
        }
    except Exception as e:
        raise HTTPException(
            status_code=503,
            detail=f"Service Unhealthy: {str(e)}"
        )

# zangao_rag.py
from langchain.vectorstores import FAISS
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from pydantic import BaseModel
from fastapi import FastAPI

app = FastAPI(title="Zangão RAG")

class QueryInput(BaseModel):
    question: str
    context: Optional[dict] = None

class RAGZangao:
    def __init__(self, knowledge_base_path):
        # 1. Carrega base de conhecimento vetorial
        self.embeddings = HuggingFaceEmbeddings()
        self.db = FAISS.load_local(knowledge_base_path, self.embeddings)
        
        # 2. Configura cadeia RAG
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=OpenAI(temperature=0),
            chain_type="stuff",
            retriever=self.db.as_retriever(search_kwargs={"k": 3}),
            return_source_documents=True
        )
    
    def query(self, question: str, context: dict = None) -> dict:
        """Processa uma consulta usando RAG"""
        augmented_question = f"Contexto: {context}\n\nPergunta: {question}" if context else question
        
        result = self.qa_chain({"query": augmented_question})
        
        return {
            "answer": result["result"],
            "sources": [doc.metadata["source"] for doc in result["source_documents"]],
            "confidence": self._calculate_confidence(result)
        }
    
    def _calculate_confidence(self, result) -> float:
        """Calcula confiança baseada na similaridade dos documentos"""
        if not result["source_documents"]:
            return 0.0
        return sum([doc.metadata.get("score", 0.5) for doc in result["source_documents"]]) / len(result["source_documents"])

# Inicialização (em produção, seria um singleton)
rag_zangao = RAGZangao("caminho/para/base_de_conhecimento")

@app.post("/query")
async def handle_query(query: QueryInput):
    return rag_zangao.query(query.question, query.context)

@app.get("/health")
async def health_check():
    return {"status": "healthy", "docs_loaded": rag_zangao.db.index.ntotal}

# operaria_com_geleia.py
import json
from datetime import datetime
from pydantic import BaseModel

class TrainingLogEntry(BaseModel):
    timestamp: datetime
    pollen_id: str
    model_version: str
    input_data: dict
    output_data: dict
    user_feedback: Optional[dict] = None
    metrics: Optional[dict] = None

def log_for_retraining(pollen_id: str, input_data: dict, 
                      output_data: dict, model_version: str):
    """Registra dados para re-treinamento futuro"""
    log_entry = TrainingLogEntry(
        timestamp=datetime.utcnow(),
        pollen_id=pollen_id,
        model_version=model_version,
        input_data=input_data,
        output_data=output_data
    )
    
    # Em produção, enviaria para um serviço de logging ou S3
    with open("training_logs.jsonl", "a") as f:
        f.write(log_entry.json() + "\n")
        
    return log_entry

# Após chamar um Zangão:
# log_for_retraining(
#     pollen_id="123",
#     input_data={"text": "..."},
#     output_data=zangao_response,
#     model_version=zangao_response["model_version"]
# )