1
🤔 Por que combinar LangGraph e CrewAI
LangGraph e CrewAI não competem — eles se complementam. LangGraph é excelente em controle de fluxo macro (quando fazer o quê, com quais condições, com checkpoints). CrewAI é excelente em execução colaborativa (equipes de agentes fazendo análise profunda). Juntos, cobrem o espectro completo.
🎯 Separação de responsabilidades
🕸️ LangGraph cuida de:
- •Ordem e condições de execução das etapas
- •Checkpoints antes e depois de cada fase
- •Lógica de retry em caso de falha
- •Human-in-the-loop quando necessário
- •Streaming de progresso para o usuário
👥 CrewAI cuida de:
- •Análise profunda em equipe de especialistas
- •Colaboração entre agentes com papéis claros
- •Output estruturado em Pydantic
- •Delegação automática de sub-tarefas
- •Geração de relatórios ricos e estruturados
💡 O princípio da combinação inteligente
Pense em LangGraph como o diretor de orquestra e CrewAI como um grupo de músicos especialistas. O diretor decide quando cada grupo toca e mantém o ritmo global. Os músicos executam a partitura com maestria dentro do seu domínio. Nenhum substitui o outro.
2
🕸️ LangGraph como orquestrador externo
No padrão de integração, LangGraph define a estrutura macro — quais etapas existem, em que ordem, com quais condições de saída. CrewAI executa dentro de nós específicos, sendo completamente opaco para o grafo externo.
🗺️ Visão do sistema integrado
START
→
Coleta de Dados
→
[Nó LangGraph]
Análise em Equipe
→
[Crew aqui!]
Validação
→
[Nó LangGraph]
Relatório Final
→
[Nó LangGraph]
END
LangGraph gerencia o fluxo completo. A Crew executa apenas na etapa de Análise — recebe dados do state, analisa em equipe, retorna resultado estruturado.
3
👥 CrewAI como nó do LangGraph
O padrão de implementação: uma função de nó do LangGraph que instancia uma Crew, chama crew.kickoff() com inputs do state, e retorna o resultado de volta para o state.
🔨 Implementação do nó de Crew
from langgraph.graph import StateGraph, START, END
from crewai import Agent, Task, Crew, Process
from typing import TypedDict
# Estado do LangGraph
class DueDiligenceState(TypedDict):
empresa: str
setor: str
dados_coletados: dict
analise_crew: dict # output da Crew vai aqui
validacao: dict
relatorio_final: str
aprovado: bool
# ===== FACTORY DE CREW =====
def criar_crew_analise(empresa: str, dados: dict) -> Crew:
"""Cria e retorna uma Crew configurada para a análise."""
analista_financeiro = Agent(
role="Financial Analyst",
goal=f"Analise a saúde financeira de {empresa}",
backstory="Especialista em valuation e análise fundamentalista com 15 anos.",
llm="gpt-4o"
)
analista_risco = Agent(
role="Risk Analyst",
goal=f"Identifique riscos críticos para {empresa}",
backstory="Ex-consultor de risco com foco em M&A e due diligence.",
llm="gpt-4o"
)
analista_mercado = Agent(
role="Market Analyst",
goal=f"Avalie o posicionamento de mercado de {empresa}",
backstory="Especialista em análise competitiva e estratégia de mercado.",
llm="gpt-4o-mini"
)
task_financeira = Task(
description=f"""Analise os dados financeiros de {empresa}:
Dados: {dados}
Foco: rentabilidade, liquidez, endividamento""",
expected_output="Análise financeira com score 1-10 e principais alertas",
agent=analista_financeiro
)
task_risco = Task(
description=f"Identifique top 5 riscos para investimento em {empresa}",
expected_output="Lista de riscos com probabilidade e impacto",
agent=analista_risco,
context=[task_financeira]
)
task_mercado = Task(
description=f"Avalie posição competitiva de {empresa} no mercado",
expected_output="Análise de posicionamento com oportunidades e ameaças",
agent=analista_mercado
)
return Crew(
agents=[analista_financeiro, analista_risco, analista_mercado],
tasks=[task_financeira, task_risco, task_mercado],
process=Process.sequential,
verbose=False
)
# ===== NÓ DO LANGGRAPH QUE EXECUTA A CREW =====
def executar_analise_crew(state: DueDiligenceState) -> dict:
"""Nó do LangGraph que instancia e executa a Crew."""
empresa = state["empresa"]
dados = state["dados_coletados"]
# Cria a Crew com o contexto do state
crew = criar_crew_analise(empresa, dados)
# Executa com timeout (importante!)
try:
resultado = crew.kickoff(inputs={
"empresa": empresa,
"setor": state["setor"]
})
return {
"analise_crew": {
"resultado": resultado.raw,
"aprovado": True,
"custo_tokens": str(resultado.token_usage)
}
}
except Exception as e:
return {
"analise_crew": {
"resultado": f"Erro na análise: {str(e)}",
"aprovado": False
}
}
4
📦 Passagem de estado entre frameworks
A fronteira entre LangGraph e CrewAI é onde bugs acontecem. O nó de integração precisa converter o estado do LangGraph para o formato de input da Crew, e serializar o output da Crew de volta para o estado — com validação em ambos os lados.
🔀 Contrato de interface claro
from pydantic import BaseModel
# Contrato de INPUT para a Crew
class CrewInput(BaseModel):
empresa: str
setor: str
dados_financeiros: dict
periodo_analise: str = "últimos 3 anos"
# Contrato de OUTPUT da Crew
class CrewOutput(BaseModel):
score_financeiro: float # 0.0 a 10.0
principais_riscos: list[str]
oportunidades: list[str]
recomendacao: str # "INVESTIR" | "AGUARDAR" | "RECUSAR"
confianca: float # 0.0 a 1.0
# Nó com conversão e validação explícitas
def executar_analise_crew(state: DueDiligenceState) -> dict:
# 1. Converte state → CrewInput com validação
try:
crew_input = CrewInput(
empresa=state["empresa"],
setor=state["setor"],
dados_financeiros=state["dados_coletados"].get("financeiros", {}),
)
except Exception as e:
return {"analise_crew": {"erro": f"Input inválido: {e}", "aprovado": False}}
# 2. Executa a Crew
crew = criar_crew_analise(crew_input.empresa, crew_input.model_dump())
resultado = crew.kickoff()
# 3. Valida e converte output → state
try:
# Task final com output_pydantic=CrewOutput
analise = resultado.pydantic # CrewOutput tipado
return {
"analise_crew": {
"score": analise.score_financeiro,
"riscos": analise.principais_riscos,
"recomendacao": analise.recomendacao,
"aprovado": analise.recomendacao != "RECUSAR"
}
}
except Exception as e:
return {"analise_crew": {"raw": resultado.raw, "erro": str(e)}}⚠️ Armadilhas comuns de integração
- ✗Passar objetos não serializáveis entre frameworks (use tipos primitivos)
- ✗Não validar o output da Crew antes de colocar no state
- ✗Criar a Crew dentro do grafo sem tratamento de erro
- ✗Sem timeout: a Crew pode demorar indefinidamente
- ✓Use try/except em todo o nó de integração
- ✓Pydantic nos dois lados: input e output da Crew
5
🔷 Padrões de integração reutilizáveis
Três padrões comprovados para combinar LangGraph e CrewAI — cada um resolve uma categoria diferente de problema.
1
Padrão Crew-no-Nó (Sequential)
Uma Crew em um único nó sequencial. O mais simples e mais comum.
grafo.add_node("coleta", coletar_dados)
grafo.add_node("analise", executar_crew) # Crew aqui
grafo.add_node("relatorio", gerar_relatorio)
grafo.add_edge(START, "coleta")
grafo.add_edge("coleta", "analise")
grafo.add_edge("analise", "relatorio")
grafo.add_edge("relatorio", END)2
Padrão Crews Paralelas (Map-Reduce)
Múltiplas Crews executando em paralelo, resultados agregados por um nó final.
from langgraph.constants import Send
# Distribui para múltiplas Crews em paralelo
def distribuir_analises(state) -> list:
return [
Send("analise_crew", {"empresa": e, **state})
for e in state["empresas_para_analisar"]
]
grafo.add_conditional_edges("preparar", distribuir_analises)
grafo.add_node("analise_crew", executar_crew) # cada empresa
grafo.add_node("agregar", combinar_resultados) # une tudo3
Padrão Loop com Crew (Retry Inteligente)
O loop verifica qualidade e chama a Crew novamente se necessário.
def checar_qualidade(state) -> str:
score = state["analise_crew"].get("score", 0)
tentativas = state.get("tentativas", 0)
if score >= 7.0 or tentativas >= 3:
return "relatorio"
return "analise_crew" # volta para a Crew
grafo.add_conditional_edges("avaliar", checar_qualidade)
6
🔍 Projeto: sistema de due diligence automatizado
O projeto completo: sistema de due diligence onde LangGraph orquestra as etapas (coleta → análise → validação → relatório) e uma CrewAI de 3 analistas executa a etapa de análise em profundidade.
🚀 Grafo completo de due diligence
from langgraph.checkpoint.memory import MemorySaver
# Nós do LangGraph (sem a Crew)
def coletar_dados(state: DueDiligenceState) -> dict:
"""Coleta dados públicos da empresa via APIs."""
empresa = state["empresa"]
return {
"dados_coletados": {
"financeiros": buscar_dados_financeiros(empresa),
"mercado": buscar_dados_mercado(empresa, state["setor"]),
"noticias": buscar_noticias_recentes(empresa),
"juridico": buscar_processos_juridicos(empresa)
}
}
def validar_analise(state: DueDiligenceState) -> dict:
"""Valida se a análise da Crew atende critérios mínimos."""
analise = state["analise_crew"]
issues = []
if analise.get("score", 0) == 0:
issues.append("Score não calculado")
if not analise.get("riscos"):
issues.append("Riscos não identificados")
return {
"validacao": {
"aprovada": len(issues) == 0,
"issues": issues
}
}
def gerar_relatorio_final(state: DueDiligenceState) -> dict:
"""Gera relatório executivo consolidado."""
analise = state["analise_crew"]
return {
"relatorio_final": f"""
# Due Diligence: {state['empresa']}
## Executive Summary
**Recomendação:** {analise.get('recomendacao', 'N/A')}
**Score Financeiro:** {analise.get('score', 'N/A')}/10
## Principais Riscos
{chr(10).join(f'- {r}' for r in analise.get('riscos', []))}
## Próximos Passos
{analise.get('proximos_passos', 'Consultar equipe jurídica')}
""",
"aprovado": analise.get("recomendacao") == "INVESTIR"
}
# Lógica de roteamento
def rotear_apos_validacao(state: DueDiligenceState) -> str:
if state["validacao"]["aprovada"]:
return "relatorio"
tentativas = state.get("tentativas", 0)
if tentativas >= 2:
return "relatorio" # força saída
return "analise" # repete a Crew
# Monta o grafo completo
grafo = StateGraph(DueDiligenceState)
grafo.add_node("coleta", coletar_dados)
grafo.add_node("analise", executar_analise_crew) # Crew aqui
grafo.add_node("validacao", validar_analise)
grafo.add_node("relatorio", gerar_relatorio_final)
grafo.add_edge(START, "coleta")
grafo.add_edge("coleta", "analise")
grafo.add_edge("analise", "validacao")
grafo.add_conditional_edges("validacao", rotear_apos_validacao, {
"relatorio": "relatorio",
"analise": "analise"
})
grafo.add_edge("relatorio", END)
# Compila com checkpoint
checkpointer = MemorySaver()
app = grafo.compile(checkpointer=checkpointer)
# Executa
config = {"configurable": {"thread_id": "dd_empresa_xyz_001"}}
resultado = app.invoke({
"empresa": "TechStartup XYZ",
"setor": "SaaS B2B",
"dados_coletados": {},
"analise_crew": {},
"validacao": {},
"relatorio_final": "",
"aprovado": False
}, config=config)
print(resultado["relatorio_final"])🎯 Resumo do Módulo
✓
LangGraph = orquestrador, CrewAI = executor — separação clara de responsabilidades entre os dois
✓
Nó de integração — função Python que instancia a Crew, executa kickoff e retorna ao state
✓
Pydantic na fronteira — valida input e output da Crew para prevenir erros de interface
✓
3 padrões — Sequential, Map-Reduce (paralelo) e Loop com Crew para diferentes necessidades
✓
Try/except obrigatório — todo nó de integração precisa tratar erros graciosamente
Próximo Módulo:
3.8 — 📊 Observabilidade: LangSmith, Arize Phoenix, métricas de negócio e avaliação contínua em produção