LangChain vs LangGraph
Arquitetura Multi-Agentes
Entenda a divisão de responsabilidades, como os grafos de estado funcionam, o padrão Supervisor/Especialista e como invocar agentes via MCP sem A2A.
A Divisão de Responsabilidades
LangChain e LangGraph não são concorrentes — são camadas complementares com responsabilidades distintas. A confusão surge porque ambos fazem parte do mesmo ecossistema e podem ser usados juntos de formas não óbvias.
- LLMs / Chat Models — abstrações sobre OpenAI, Anthropic, Bedrock
- Prompts — ChatPromptTemplate, MessagesPlaceholder
- Tools — @tool, StructuredTool, ToolNode
- Memory — ConversationBufferMemory, checkpointers
- Retrievers — integração com vector stores para RAG
- Chains — pipelines lineares com | (LCEL)
- StateGraph — grafo de estado tipado com TypedDict
- Nós — funções que lêem e escrevem no estado
- Arestas condicionais — roteamento dinâmico decidido pelo LLM
- Checkpointer — persistência de estado entre execuções
- Interrupt — human-in-the-loop nativo
- Subgrafos — composição de fluxos complexos
LangChain sozinho resolve fluxos lineares e sem estado. Quando você precisa de loops, decisões condicionais, estado persistente entre etapas e múltiplos agentes colaborando — aí entra o LangGraph.
Grafos: Como Funcionam de Verdade
Um grafo LangGraph não é um fluxograma de caixas. É uma máquina de estados onde cada transição pode ser determinística ou decidida pelo LLM em tempo de execução.
Estado (TypedDict) ↓ Nó lê o estado → processa → escreve de volta no estado ↓ Aresta decide qual nó vem a seguir (fixo ou condicional) ↓ Repete até chegar em END
Definindo o Estado — o contrato compartilhado
O Estado é o TypedDict compartilhado entre todos os nós. A chave do design é o uso de reducers via Annotated — eles definem como o estado é atualizado quando múltiplos nós escrevem na mesma chave.
from typing import TypedDict, Annotated, List
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage
class AgentState(TypedDict):
# Annotated + reducer: define COMO o campo é atualizado
# add_messages faz append em vez de sobrescrever
messages: Annotated[List[BaseMessage], add_messages]
intent: str # intenção detectada pelo nó de classificação
current_agent: str # agente ativo no momento
session_id: str # id de sessão (passa para tools MCP)
context: dict # dados acumulados na sessão
final_answer: str # resposta final para o usuário
add_messages acumula mensagens sem perder histórico. Para outros campos que precisam acumular, defina seu próprio reducer.
Tipos de Arestas
# Aresta fixa — sempre vai para o mesmo nó
graph.add_edge("node_a", "node_b")
# Aresta condicional — função decide o próximo nó em runtime
graph.add_conditional_edges(
"supervisor",
route_decision, # retorna nome do próximo nó
{
"agent_pesquisa": "agent_pesquisa",
"agent_dados": "agent_dados",
"FINISH": END
}
)
Padrão Supervisor + Agentes Especialistas
┌─────────────────────────────────────────────────┐
│ SUPERVISOR │
│ - Detecta intenção │
│ - Decide qual agente invocar via tool_calls │
│ - Agrega respostas no estado │
│ - Decide quando terminar (END) │
└────────┬──────────────┬──────────────┬──────────┘
│ │ │
┌───────▼───┐ ┌───────▼───┐ ┌──────▼──────┐
│ Agente │ │ Agente │ │ Agente │
│ Pesquisa │ │ Análise │ │ Relatório │
└───────────┘ └───────────┘ └─────────────┘
O Supervisor é um nó LLM com prompt que instrui o modelo a escolher qual agente invocar. Ele não executa tarefas — apenas decide e delega.
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_anthropic import ChatAnthropic
supervisor_prompt = ChatPromptTemplate.from_messages([
("system", """Você é o supervisor de agentes.
Com base na conversa, decida qual agente especialista invocar.
Agentes disponíveis:
- agent_pesquisa : busca e coleta informações
- agent_analise : analisa dados e identifica padrões
- agent_relatorio : formata e gera relatórios
- FINISH : quando a tarefa estiver completa
Responda chamando a tool do agente correto."""),
MessagesPlaceholder(variable_name="messages"),
])
llm = ChatAnthropic(model="claude-sonnet-4-20250514")
def supervisor_node(state: AgentState):
mensagens = trim_messages(
state["messages"],
strategy="last",
token_counter=llm,
max_tokens=8000,
start_on="human",
include_system=True,
)
response = (supervisor_prompt | llm.bind_tools(tools)).invoke(mensagens)
return {"messages": [response]}
Loop Supervisor ↔ ToolNode
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
def should_continue(state: AgentState) -> str:
last = state["messages"][-1]
# Se o LLM chamou alguma tool → vai executar
if last.tool_calls:
return "tools"
# Nenhum tool_call → supervisor decidiu terminar
return END
tool_node = ToolNode(tools) # executa qualquer tool chamada pelo LLM
graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("supervisor")
graph.add_conditional_edges(
"supervisor", should_continue,
{"tools": "tools", END: END}
)
# Após executar a tool, volta pro supervisor decidir o próximo passo
graph.add_edge("tools", "supervisor")
app = graph.compile(checkpointer=checkpointer)
Agent-as-a-Tool via MCP
Sem A2A, cada agente especialista é exposto como uma tool MCP e invocado pelo supervisor via tool_calls. Para o LangGraph, é indiferente se a tool é uma função local ou um agente completo rodando em outro servidor — a interface é a mesma.
Supervisor (LangGraph local) │ ├── tool: mcp_agent_pesquisa() → MCP Server → ECS Agent Pesquisa ├── tool: mcp_agent_analise() → MCP Server → ECS Agent Análise └── tool: mcp_agent_relatorio() → MCP Server → ECS Agent Relatório
Carregando tools do MCP
from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient({
"agente_pesquisa": {
"url": "https://mcp-agentes.interno/mcp",
"transport": "streamable_http"
}
})
# Cada tool MCP vira um objeto Tool do LangChain
# O supervisor as usa exatamente como qualquer outra tool
tools = await client.get_tools()
Fluxo de execução com MCP
Intenções e Fluxos de Definição dos Grafos
Em vez de deixar o supervisor detectar intenção implicitamente, o padrão recomendado é um nó dedicado de classificação como primeiro passo do grafo. Isso torna o roteamento previsível e auditável.
from enum import Enum
class Intent(str, Enum):
PESQUISA_SIMPLES = "pesquisa_simples"
ANALISE_DADOS = "analise_dados"
RELATORIO = "relatorio"
CONVERSA = "conversa"
DESCONHECIDO = "desconhecido"
intent_prompt = ChatPromptTemplate.from_messages([
("system", """Classifique a intenção em UMA categoria:
- pesquisa_simples : quer buscar informações
- analise_dados : quer analisar ou comparar dados
- relatorio : quer um documento formatado
- conversa : conversa casual, sem tarefa
- desconhecido : não foi possível classificar
Responda APENAS com o nome da categoria."""),
MessagesPlaceholder("messages")
])
def intent_node(state: AgentState) -> dict:
response = (intent_prompt | llm).invoke(state)
return {"intent": response.content.strip()}
def route_by_intent(state: AgentState) -> str:
routing_map = {
Intent.PESQUISA_SIMPLES: "supervisor",
Intent.ANALISE_DADOS: "supervisor",
Intent.RELATORIO: "supervisor",
Intent.CONVERSA: "chat_simples",
Intent.DESCONHECIDO: "clarification",
}
return routing_map.get(state["intent"], "clarification")
Grafo completo com intenções
graph = StateGraph(AgentState)
graph.add_node("detect_intent", intent_node)
graph.add_node("supervisor", supervisor_node)
graph.add_node("tools", tool_node)
graph.add_node("chat_simples", chat_node)
graph.add_node("clarification", clarification_node)
# Entrada sempre pelo detector de intenção
graph.set_entry_point("detect_intent")
# Roteamento por intenção
graph.add_conditional_edges("detect_intent", route_by_intent, {
"supervisor": "supervisor",
"chat_simples": "chat_simples",
"clarification": "clarification",
})
# Loop supervisor ↔ tools
graph.add_conditional_edges(
"supervisor", should_continue,
{"tools": "tools", END: END}
)
graph.add_edge("tools", "supervisor")
# Terminações diretas
graph.add_edge("chat_simples", END)
graph.add_edge("clarification", END)
app = graph.compile(checkpointer=MemorySaver())