Orquestração Agêntica com LangGraph: Do Zero ao Deploy
Orquestração Agêntica com LangGraph: Do Zero ao Deploy
Agentes de IAFeatured

Orquestração Agêntica com LangGraph: Do Zero ao Deploy

Como construir pipelines de agentes multi-step com estado persistente usando LangGraph, LangSmith e FastAPI. Um guia prático com código real de produção.

M
Moises CostaDecember 10, 202512 min read
LangGraphLangChainPythonFastAPIAgents

Orquestração Agêntica com LangGraph: Do Zero ao Deploy

Construir agentes de IA que realmente funcionam em produção é um desafio diferente de criar demos. Neste artigo, vou mostrar como uso LangGraph para orquestrar pipelines agênticos complexos com estado persistente, checkpointing e deploy via FastAPI.

O Problema com Agentes Simples

A maioria dos tutoriais de agentes de IA mostra um loop básico: LLM recebe pergunta → chama ferramenta → retorna resposta. Isso funciona para demos, mas falha em produção por várias razões:

  • Sem estado persistente: cada chamada começa do zero
  • Sem controle de fluxo: o LLM decide tudo, inclusive quando parar
  • Sem observabilidade: você não sabe o que aconteceu quando algo falha
  • Sem recuperação de erros: uma falha derruba todo o pipeline

O LangGraph resolve todos esses problemas com uma abstração elegante: grafos de estado.

Conceitos Fundamentais do LangGraph

StateGraph

O StateGraph é o coração do LangGraph. Você define um estado tipado (usando TypedDict ou Pydantic) e nós que transformam esse estado:

python
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class AgentState(TypedDict):
    messages: Annotated[List[dict], operator.add]
    current_step: str
    tool_results: List[dict]
    final_answer: str | None

graph = StateGraph(AgentState)

Nós e Arestas

Cada nó é uma função Python que recebe e retorna o estado:

python
def call_llm(state: AgentState) -> AgentState:
    messages = state["messages"]
    response = llm.invoke(messages)
    return {
        "messages": [{"role": "assistant", "content": response.content}],
        "current_step": "check_tools"
    }

def execute_tools(state: AgentState) -> AgentState:
    # Executa ferramentas baseado na resposta do LLM
    tool_calls = extract_tool_calls(state["messages"][-1])
    results = []
    for call in tool_calls:
        result = tools[call["name"]](**call["args"])
        results.append({"tool": call["name"], "result": result})
    return {"tool_results": results, "current_step": "synthesize"}

# Adiciona nós ao grafo
graph.add_node("llm", call_llm)
graph.add_node("tools", execute_tools)

Roteamento Condicional

O poder real vem do roteamento condicional — o grafo decide qual caminho seguir baseado no estado:

python
def should_use_tools(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if has_tool_calls(last_message):
        return "tools"
    return END

graph.add_conditional_edges("llm", should_use_tools)
graph.add_edge("tools", "llm")  # Volta para o LLM após executar ferramentas
graph.set_entry_point("llm")

Persistência com Checkpointing

Um dos recursos mais poderosos do LangGraph é o checkpointing. Cada execução pode ser salva e retomada:

python
from langgraph.checkpoint.sqlite import SqliteSaver

# Em produção, use PostgreSQL
checkpointer = SqliteSaver.from_conn_string("./checkpoints.db")
app = graph.compile(checkpointer=checkpointer)

# Executa com um thread_id único para persistência
config = {"configurable": {"thread_id": "user-123-session-456"}}
result = app.invoke({"messages": [{"role": "user", "content": "Analise este contrato"}]}, config)

# Retoma a conversa mais tarde
result2 = app.invoke({"messages": [{"role": "user", "content": "Agora compare com o anterior"}]}, config)

Deploy com FastAPI e Streaming

Para produção, exponho o agente via FastAPI com streaming SSE:

python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

@app.post("/agent/stream")
async def stream_agent(request: AgentRequest):
    async def generate():
        async for event in agent_app.astream_events(
            {"messages": [{"role": "user", "content": request.message}]},
            config={"configurable": {"thread_id": request.session_id}},
            version="v2"
        ):
            if event["event"] == "on_chat_model_stream":
                chunk = event["data"]["chunk"].content
                if chunk:
                    yield f"data: {json.dumps({'token': chunk})}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")

Observabilidade com LangSmith

Integro LangSmith para rastrear cada execução em produção:

python
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "sua-chave-aqui"
os.environ["LANGCHAIN_PROJECT"] = "meu-agente-producao"

Com isso, cada execução fica registrada no LangSmith com: latência por nó, tokens consumidos, erros e o grafo de execução completo.

Padrões de Produção

Após 2 anos usando LangGraph em produção no Detran-RJ, aprendi alguns padrões essenciais:

1. Sempre defina timeouts por nó:

python
from langgraph.graph import StateGraph
import asyncio

async def call_llm_with_timeout(state):
    try:
        return await asyncio.wait_for(call_llm(state), timeout=30.0)
    except asyncio.TimeoutError:
        return {"error": "LLM timeout", "current_step": "error_handler"}

2. Implemente circuit breakers para ferramentas externas:

python
from functools import wraps

def with_fallback(fallback_value):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            try:
                return await func(*args, **kwargs)
            except Exception as e:
                logger.error(f"Tool failed: {e}")
                return fallback_value
        return wrapper
    return decorator

3. Use sub-grafos para modularidade: Separe responsabilidades em sub-grafos independentes que podem ser testados e deployados separadamente.

Conclusão

LangGraph transformou como construo agentes de IA. A combinação de estado tipado, checkpointing e roteamento condicional permite criar sistemas robustos que realmente funcionam em produção. O overhead inicial de aprender a API vale muito a pena.

No próximo artigo, vou mostrar como combino LangGraph com RAG para criar agentes que consultam bases de conhecimento específicas do domínio.

Moises Costa é AI Engineer no Detran-RJ e criador da MSc Academy. Conecte-se no LinkedIn.

References & Resources

Live AI Agents

See the concepts in action

Interact with Arquimedes, Atlas, Artemis and the Detran-RJ agent — built with the exact techniques described in this article.

Related Articles

RAG em Produção: Arquitetura, Chunking e Avaliação
RAG & Vector DBs

RAG em Produção: Arquitetura, Chunking e Avaliação

18 min
MCP (Model Context Protocol): O Futuro da Integração de Agentes
Agentes de IA

MCP (Model Context Protocol): O Futuro da Integração de Agentes

8 min
Full Stack AI: React + FastAPI + LangChain em Produção
Full Stack AI

Full Stack AI: React + FastAPI + LangChain em Produção

20 min