👥 CrewAI na Prática

from crewai import Agent
from crewai_tools import SerperDevTool, WebsiteSearchTool

search_tool = SerperDevTool()

# Agente 1: Pesquisador de Mercado
pesquisador = Agent(
    role="Senior Market Research Analyst",
    goal="Coletar dados precisos e atualizados sobre tendências de mercado no setor solicitado",
    backstory="""Você é um analista sênior com 10 anos de experiência em pesquisa
    de mercado. Especialista em identificar padrões emergentes, analisar dados
    qualitativos e quantitativos, e sintetizar informações complexas em insights
    acionáveis. Sua metodologia é rigorosa e suas fontes são sempre verificadas.""",
    tools=[search_tool],
    llm="gpt-4o",
    verbose=True,
    allow_delegation=False  # não pode delegar sub-tarefas
)

# Agente 2: Analista Estratégico
analista = Agent(
    role="Strategic Business Analyst",
    goal="Transformar dados brutos em análises estratégicas com recomendações claras",
    backstory="""Especialista em análise SWOT, Porter's Five Forces e frameworks
    estratégicos. Você transforma dados e pesquisas em narrativas coerentes que
    executivos conseguem usar para tomar decisões. Seu diferencial é conectar
    pontos que outros não veem.""",
    llm="gpt-4o",
    verbose=True
)

# Agente 3: Redator Executivo
redator = Agent(
    role="Executive Content Writer",
    goal="Criar relatórios executivos claros, concisos e persuasivos a partir de análises",
    backstory="""Redator especializado em comunicação executiva. Você sabe como
    apresentar dados complexos de forma que C-level entenda em 5 minutos.
    Usa estrutura clara: executive summary, principais achados, recomendações
    e próximos passos.""",
    llm="gpt-4o-mini",  # redação usa modelo menor
    verbose=True
)

from crewai import Task
from pydantic import BaseModel
from typing import List

# Schema do output final
class RelatorioMercado(BaseModel):
    setor: str
    principais_tendencias: List[str]
    oportunidades: List[str]
    ameacas: List[str]
    recomendacoes: List[str]
    score_atratividade: float

# Task 1: Pesquisa
task_pesquisa = Task(
    description="""Pesquise o mercado de {setor} focando em:
    1. Tamanho de mercado atual e projeções 2025-2027
    2. Top 5 players e suas estratégias
    3. Tendências tecnológicas emergentes
    4. Barreiras de entrada
    5. Comportamento do consumidor""",
    expected_output="""Relatório de pesquisa com dados quantitativos
    de pelo menos 5 fontes verificadas. Include: tamanho de mercado (USD),
    CAGR, principais players com market share, e 3 tendências com evidências.""",
    agent=pesquisador,
    output_file="pesquisa_raw.md"  # salva em arquivo
)

# Task 2: Análise (usa output da pesquisa como contexto)
task_analise = Task(
    description="""Com base na pesquisa de mercado fornecida, realize:
    1. Análise SWOT completa do setor
    2. Mapeamento de oportunidades não óbvias
    3. Avaliação de riscos com probabilidade e impacto
    4. Score de atratividade do mercado (0-10)""",
    expected_output="""Análise estratégica estruturada com SWOT,
    mapa de oportunidades rankeadas por potencial, matriz de risco
    e score de atratividade com justificativa.""",
    agent=analista,
    context=[task_pesquisa]  # recebe output da task anterior
)

# Task 3: Redação (output em Pydantic)
task_relatorio = Task(
    description="""Crie um relatório executivo baseado na pesquisa e análise.
    O relatório deve ser legível em 5 minutos por um CEO.""",
    expected_output="Relatório executivo em formato JSON estruturado.",
    agent=redator,
    context=[task_pesquisa, task_analise],
    output_pydantic=RelatorioMercado  # output tipado!
)

from crewai import Crew, Process

# Manager agent (usa modelo forte para coordenação)
manager = Agent(
    role="Research Project Manager",
    goal="Garantir que a pesquisa de mercado seja completa, precisa e entregue no prazo",
    backstory="Gerente de projetos com experiência em coordenar times de análise.",
    llm="gpt-4o",
    allow_delegation=True  # DEVE ter delegation ativo
)

# Crew com hierarquia
crew_hierarquica = Crew(
    agents=[pesquisador, analista, redator],
    tasks=[task_pesquisa, task_analise, task_relatorio],
    process=Process.hierarchical,
    manager_agent=manager,
    verbose=True,
    memory=True  # memória compartilhada entre agentes
)

resultado = crew_hierarquica.kickoff(
    inputs={"setor": "IA Generativa B2B"}
)

# Acessa output tipado
relatorio: RelatorioMercado = resultado.pydantic
print(f"Score de atratividade: {relatorio.score_atratividade}")
print(f"Recomendações: {relatorio.recomendacoes}")

# Cenário: pesquisa de 3 mercados simultaneamente
task_mercado_br = Task(
    description="Analise o mercado de IA no Brasil",
    expected_output="Relatório mercado Brasil",
    agent=pesquisador,
    async_execution=True  # roda em paralelo!
)

task_mercado_us = Task(
    description="Analise o mercado de IA nos EUA",
    expected_output="Relatório mercado EUA",
    agent=pesquisador,
    async_execution=True  # também paralelo!
)

task_mercado_eu = Task(
    description="Analise o mercado de IA na Europa",
    expected_output="Relatório mercado Europa",
    agent=pesquisador,
    async_execution=True  # também paralelo!
)

# Esta task espera todos os anteriores terminarem
task_comparativo = Task(
    description="Compare os três mercados e identifique o mais atrativo",
    expected_output="Comparativo global com ranking",
    agent=analista,
    context=[task_mercado_br, task_mercado_us, task_mercado_eu],
    # sem async_execution — espera o contexto
)

# Crew executa: 3 pesquisas em paralelo, depois o comparativo
crew = Crew(
    agents=[pesquisador, analista],
    tasks=[task_mercado_br, task_mercado_us, task_mercado_eu, task_comparativo],
    process=Process.sequential
)

# Crew completa de análise de mercado
from crewai import Crew, Process
from datetime import datetime

crew_analise = Crew(
    agents=[pesquisador, analista, redator],
    tasks=[task_pesquisa, task_analise, task_relatorio],
    process=Process.sequential,
    verbose=True,
    memory=True,           # memória entre tasks
    max_rpm=10,            # rate limiting de API
    output_log_file=f"log_{datetime.now():%Y%m%d_%H%M}.txt"
)

# Executa com inputs dinâmicos
resultado = crew_analise.kickoff(inputs={
    "setor": "SaaS de Recursos Humanos",
    "regiao": "América Latina",
    "foco": "PMEs com 50-500 funcionários"
})

# Output estruturado e tipado
relatorio = resultado.pydantic
print(f"Setor analisado: {relatorio.setor}")
print(f"Score de atratividade: {relatorio.score_atratividade}/10")
print("\nPrincipais tendências:")
for i, tendencia in enumerate(relatorio.principais_tendencias, 1):
    print(f"  {i}. {tendencia}")
print("\nRecomendações estratégicas:")
for rec in relatorio.recomendacoes:
    print(f"  → {rec}")

# Custos e métricas da execução
print(f"\nTokens usados: {resultado.token_usage}")
print(f"Tempo total: {resultado.tasks_output[-1].raw[:100]}...")