06-04-01 — CrewAI e Pydantic AI

CrewAI: role-based multi-agent

A metáfora do CrewAI é uma crew (equipe): cada membro tem um papel, objetivos e comportamentos definidos. A crew trabalha junta para completar uma missão.

Conceitos principais

• Agent: entidade com role, goal, backstory e tools
• Task: unidade de trabalho com description, expected_output e agente responsável
• Crew: agrupa agents e tasks, define processo (sequential ou hierarchical)
• Process: como as tasks são executadas — sequential (uma a uma) ou hierarchical (manager delega)

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool

search_tool = SerperDevTool()

# Define a equipe
pesquisador = Agent(
    role="Pesquisador de Mercado Senior",
    goal="Coletar dados precisos sobre o mercado de IA no Brasil",
    backstory="""Especialista em análise competitiva com 10 anos de experiência.
    Você é meticuloso e sempre cita fontes verificáveis.""",
    tools=[search_tool],
    llm="gpt-4o-mini",
    verbose=True
)

analista = Agent(
    role="Analista de Negócios",
    goal="Transformar dados de mercado em insights acionáveis",
    backstory="Especialista em traduzir dados técnicos em linguagem executiva.",
    llm="gpt-4o",
    verbose=True
)

escritor = Agent(
    role="Redator Técnico",
    goal="Produzir relatórios executivos claros e persuasivos",
    backstory="Especialista em comunicação corporativa para C-level.",
    llm="gpt-4o-mini",
    verbose=True
)

# Define as tarefas
tarefa_pesquisa = Task(
    description="Pesquise o mercado de AI consulting no Brasil em 2026. Foco em: tamanho de mercado, principais players, crescimento esperado.",
    expected_output="Relatório com 5-7 dados quantitativos e 3 tendências principais.",
    agent=pesquisador
)

tarefa_analise = Task(
    description="Analise os dados coletados e identifique oportunidades para uma consultoria .NET especializada em IA.",
    expected_output="3 oportunidades concretas com potencial de receita estimado.",
    agent=analista,
    context=[tarefa_pesquisa]  # depende da tarefa anterior
)

tarefa_relatorio = Task(
    description="Escreva um relatório executivo de 1 página baseado na análise.",
    expected_output="Relatório executivo em PT-BR, estilo McKinsey, máximo 500 palavras.",
    agent=escritor,
    context=[tarefa_pesquisa, tarefa_analise]
)

# Monta e executa a crew
crew = Crew(
    agents=[pesquisador, analista, escritor],
    tasks=[tarefa_pesquisa, tarefa_analise, tarefa_relatorio],
    process=Process.sequential,
    verbose=True,
    memory=True  # habilita memória entre execuções
)

resultado = crew.kickoff()
print(resultado.raw)

Processo Hierarchical

# No processo hierarchical, um manager LLM delega automaticamente
crew_hierarquica = Crew(
    agents=[pesquisador, analista, escritor],
    tasks=[tarefa_complexa],  # uma task genérica — manager decide como dividir
    process=Process.hierarchical,
    manager_llm="gpt-4o",  # LLM que atua como gerente
    verbose=True
)

Pydantic AI: type-safe agents

Criado pelo time do Pydantic (autores da biblioteca de validação Python mais usada), Pydantic AI parte de uma premissa: agentes em produção precisam retornar dados estruturados e validados, não strings livres.

Conceitos principais

• Agent[OutputType]: agente genérico que garante o tipo de saída
• RunContext[DependencyType]: dependências injetadas (banco, APIs) com type safety
• @agent.tool: decorator para registrar ferramentas
• result.data: sempre um objeto Python tipado

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.openai import OpenAIModel
from dataclasses import dataclass

# Define saída estruturada
class AnaliseContrato(BaseModel):
    cliente: str
    valor_total: float
    data_vencimento: str
    clausulas_risco: list[str]
    recomendacao: str
    risco_score: int  # 1-10

# Define dependências injetadas
@dataclass
class ContratoDeps:
    db_connection: str  # em produção: AsyncConnection
    empresa: str

# Cria o agente tipado
agente = Agent(
    model=OpenAIModel("gpt-4o-mini"),
    result_type=AnaliseContrato,  # saída SEMPRE será AnaliseContrato
    deps_type=ContratoDeps,
    system_prompt="""Você é um especialista jurídico em análise de contratos.
    Sempre retorne uma análise estruturada e objetiva."""
)

# Ferramenta com acesso às dependências
@agente.tool
async def buscar_historico_cliente(ctx: RunContext[ContratoDeps], cliente: str) -> str:
    """Busca histórico de contratos anteriores do cliente."""
    # ctx.deps.db_connection disponível aqui
    return f"Cliente {cliente}: 3 contratos anteriores, todos cumpridos."

# Executar — resultado GARANTIDAMENTE tipado
async def main():
    deps = ContratoDeps(db_connection="postgresql://...", empresa="Vale")
    
    result = await agente.run(
        "Analise o contrato: Vale, R$2M, vencimento 2027-12, cláusula de reajuste anual pelo IPCA",
        deps=deps
    )
    
    analise: AnaliseContrato = result.data  # type checker sabe que é AnaliseContrato
    print(f"Risco: {analise.risco_score}/10")
    print(f"Recomendação: {analise.recomendacao}")

# Streaming com output estruturado
async def com_streaming():
    async with agente.run_stream("Analise o contrato...", deps=deps) as stream:
        async for chunk in stream.stream_text():
            print(chunk, end="")
        result = await stream.get_data()  # AnaliseContrato completo no final

Testabilidade — diferencial do Pydantic AI

from pydantic_ai import models

# Modo teste: sem chamadas reais ao LLM
with models.override(model=models.TestModel()):
    result = await agente.run("Analise este contrato")
    assert isinstance(result.data, AnaliseContrato)

# TestModel retorna dados válidos conforme o schema — ideal para CI/CD

CrewAI vs Pydantic AI — quando usar cada um