Por que Multi-Agentes LangGraph Quebram em Produção e Como Recuperá-los no Nível de Código

Existe um equívoco comum entre desenvolvedores backend que estão migrando de cadeias de prompt únicas para sistemas multi-agentes: a ideia de que escrever prompts melhores deixará o sistema estável. A maioria dos problemas que ocorrem em produção não tem nada a ver com o prompt. As verdadeiras causas são problemas de arquitetura de sistema, como contaminação de estado, loops infinitos, API Rate Limits e traces assíncronos impossíveis de debugar.

Para colocar um sistema multi-agentes baseado em LangGraph em ambiente de produção, é necessário tratar o sistema não como um conjunto de recursos de prompt, mas sim como um sistema backend, lidando com isolamento de estado, controle de concorrência e tracing diretamente no nível de código.

Isolamento de Contaminação de Estado e Bloqueio de Loops Infinitos em Nós Concorrentes

Em um grafo baseado em estado, se múltiplos nós modificarem diretamente um único objeto compartilhado, ocorrerá uma condição de corrida (race condition). Em um padrão de fan-out com execução concorrente, se você sobrescrever campos comuns sem um redutor (reducer) dedicado, apenas o resultado do nó que terminar por último será preservado, e o restante será perdido.

Para evitar isso, é necessário anexar um redutor explícito ao estado do grafo principal e encapsular totalmente os agentes secundários como subgrafos com esquemas independentes.

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

class ParentState(TypedDict):
task_id: str
input_query: str
audit_logs: Annotated[List[str], operator.add]
final_response: str

class InternalAgentState(TypedDict):
sub_task: str
scratchpad_messages: List[str]
sub_result: str

def internal_processing_node(state: InternalAgentState) -> dict:
updated_messages = state["scratchpad_messages"] + ["내부 격리 추론 진행 중"]
return {
"scratchpad_messages": updated_messages,
"sub_result": f"하위 작업 완료: {state['sub_task']}"
}

subgraph_builder = StateGraph(InternalAgentState)
subgraph_builder.add_node("internal_processing", internal_processing_node)
subgraph_builder.add_edge(START, "internal_processing")
subgraph_builder.add_edge("internal_processing", END)
compiled_subgraph = subgraph_builder.compile()

def call_isolated_subgraph_wrapper(state: ParentState) -> dict:
subgraph_input: InternalAgentState = {
"sub_task": state["input_query"],
"scratchpad_messages": []
}
subgraph_output = compiled_subgraph.invoke(subgraph_input)
return {
"audit_logs": [f"[서브그래프 결과]: {subgraph_output['sub_result']}"]
}

parent_builder = StateGraph(ParentState)
parent_builder.add_node("isolated_agent", call_isolated_subgraph_wrapper)
parent_builder.add_edge(START, "isolated_agent")
parent_builder.add_edge("isolated_agent", END)
main_graph = parent_builder.compile()

No ParentState principal, definimos o redutor operator.add para o campo audit_logs, onde ocorrem as escritas paralelas. As tarefas secundárias são isoladas em um subgrafo que usa seu próprio estado, o InternalAgentState, e os dados são trocados exclusivamente por meio de uma função wrapper. Ao bloquear a contaminação de dados, é possível reduzir o tempo de depuração de loops infinitos em mais de 5 horas por semana.

Outro problema comum é o loop de feedback do ReAct entrar em rotação infinita por não satisfazer as condições de encerramento. É necessário um roteador de guardrail que mantenha um contador no esquema de estado e o filtre nas arestas condicionais.

`python
from typing import Literal, TypedDict
from langgraph.graph import END, START, StateGraph

class GuardedState(TypedDict):
query: str
draft: str
feedback: str
is_approved: bool
iterations: int
max_iterations: int

def drafting_node(state: GuardedState) -> dict:
return {
"draft": f"작성된 초안 (반복 회차: {state['iterations'] + 1})",
"iterations": state["iterations"] + 1
}

def review_node(state: GuardedState) -> dict:
approved = state["iterations"] >= 3
return {
"is_approved": approved,
"feedback": "승인 완료" if approved else "반려: 내용 수정 필요"
}

def loop_guardrail_router(state: GuardedState) -> Literal["drafting", "fallback_escalation", "end"]:
if state["is_approved"]:
return END
if state["iterations"] >= state["max_iterations"]:
return "fallback_escalation"
return "drafting"

def fallback_escalation_node(state: GuardedState) -> dict:
return {
"draft": "에이전트 검수 피드백 루프 최대 횟수 초과. 담당자 수동 검토 건으로 이관 처리되었습니다."
}

builder = StateGraph(GuardedState)
builder.add_node("drafting", drafting_node)
builder.add_node("review", review_node)
builder.add_node("fallback_escalation", fallback_escalation_node)
builder.add_edge(START, "drafting")
builder.add_edge("drafting", "review")
builder.add_conditional_edges(
"review",
loop_guardrail_router,
{
"drafting": "drafting",
"fallback_escalation": "fallback_escalation",
END: END
}
)
builder.add_edge("fallback_escalation", END)
guarded_graph = builder.compile()

Configuramos o fluxo para desviar imediatamente para um nó de transbordo manual (fallback_escalation) quando o contador (iterations) atinge o limite configurado (max_iterations). Isso interrompe de forma definitiva o desperdício de tokens provocado por ciclos infinitos.

Prevenindo Erros HTTP 429 com Controle de Concorrência

Disparar nós secundários em paralelo de uma só vez atinge o limite de tokens por minuto (TPM) das APIs da OpenAI ou Anthropic, resultando em erros HTTP 429. No momento em que o backend oscila, toda a transação fica paralisada.

É preciso limitar o número de requisições concorrentes usando asyncio.Semaphore e aplicar Backoff Exponencial com a biblioteca tenacity para evitar que a API caia.

`python
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

API_SEMAPHORE = asyncio.Semaphore(5)

@retry(
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=2, max=10),
retry=retry_if_exception_type(Exception),
reraise=True
)
async def safe_llm_call_with_backoff(llm: ChatOpenAI, prompt: str) -> str:
async with API_SEMAPHORE:
response = await llm.ainvoke([HumanMessage(content=prompt)])
return response.content

async def parallel_worker_node(state: dict) -> dict:
llm = ChatOpenAI(model="gpt-4o", temperature=0)
task_input = state["task_data"]
result_text = await safe_llm_call_with_backoff(llm, f"하위 작업 처리: {task_input}")
return {"results": [result_text]}

Limitamos as chamadas simultâneas a no máximo 5 e, em caso de falha, realizamos tentativas com tempo de espera dobrado, variando de 2 a 10 segundos. Isso previne completamente interrupções no sistema causadas por falhas em chamadas de API externas.

Em uma abordagem de loop único, se apenas um nó falhar, é necessário gastar mais de 14 segundos para re-inferir tudo. No entanto, ao isolar os nós e aplicar backoff dessa forma, o tempo de recuperação de falhas pode ser reduzido para cerca de 10 ms. Como os resultados dos nós bem-sucedidos são mantidos, não há consumo desnecessário e repetido de tokens.

Tracing Distribuído e Retomada de Checkpoints com PostgresSaver

Agentes entrelaçados assincronamente não podem ter seu fluxo compreendido apenas com saídas no console. É necessário conectar plataformas de observabilidade baseadas em OpenTelemetry, como o Langfuse, e incluir no span de tracing até mesmo lógicas que não envolvem LLM, como operações de banco de dados e processamento de backend, para identificar gargalos.

`python
import os
from langfuse.decorators import observe, langfuse_context
from langgraph.graph import StateGraph, START, END

@observe(name="vector_store_retrieval")
def query_vector_store(query: str) -> list:
langfuse_context.update_current_observation(
input={"query": query},
metadata={"top_k": 3, "database": "pgvector"}
)
return ["문서 1: 보안 규정 예시", "문서 2: 서비스 약관"]

def retrieval_node(state: dict) -> dict:
docs = query_vector_store(state["query"])
return {"context": docs}

def execute_graph_with_tracing(app, user_query: str, session_id: str, user_id: str):
from langfuse.callback import CallbackHandler
langfuse_handler = CallbackHandler()
config = {
"configurable": {"thread_id": session_id},
"callbacks": [langfuse_handler]
}
return app.invoke({"query": user_query}, config=config)

Usando o decorador @observe, incluímos funções comuns, como a busca vetorial (vector search), no trace. Ao passar o CallbackHandler durante a chamada do grafo, é possível visualizar todo o comportamento dos agentes e o consumo de tokens por sessão em um só lugar.

Se ocorrer um erro no 10º nó no meio da execução, executar tudo do zero faz com que dinheiro e tempo sejam jogados fora simultaneamente. Usando o checkpointer PostgresSaver, os snapshots de execução dos nós ficam salvos diretamente no banco de dados.

`python
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import StateGraph

DATABASE_URL = "postgresql://postgres:postgres@localhost:5432/agent_checkpoints"

def build_app_graph():
builder = StateGraph(dict)
return builder

def resume_execution_from_failure(app, thread_id: str, fixed_payload: dict, last_valid_node: str):
config = {"configurable": {"thread_id": thread_id}}

app.update_state(
    config,
    values=fixed_payload,
    as_node=last_valid_node
)

resumed_output = app.invoke(None, config)
return resumed_output

Após verificar o último estado válido com get_state_history, corrigimos os dados problemáticos com update_state e chamamos app.invoke(None, config). O processo é retomado exatamente a partir do ponto onde havia parado.

Redução de 40% nos Custos de Tokens com Mix de Modelos e Cache Semântico

Usar o GPT-4o em todos os nós de agentes é um desperdício de orçamento. A técnica de tiering, que utiliza modelos de alto desempenho para o planejamento principal e conecta modelos leves, como o Claude 3.5 Haiku, a nós de classificação simples ou auditoria, é fundamental.

Adicionando a isso um cache semântico baseado em RedisVL, solicitações de auditoria idênticas ou semelhantes retornam em apenas 50 ms sem sequer chamar o LLM.

`python
from redisvl.extensions.llmcache import SemanticCache
from langchain_community.chat_models import ChatAnthropic

audit_semantic_cache = SemanticCache(
name="audit_nodes_cache",
redis_url="redis://localhost:6379",
distance_threshold=0.1,
ttl=86400
)

def audit_verification_node(state: dict) -> dict:
prompt_query = f"다음 최종 결과물의 정책 준수 여부를 검수하세요: {state['final_response']}"

cached_response = audit_semantic_cache.check(prompt=prompt_query)
if cached_response:
    return {
        "audit_passed": cached_response[0]["response"] == "PASSED",
        "audit_logs": ["[Audit Node]: 시맨틱 캐시 데이터 활용 (LLM 호출 스킵)"]
    }

audit_llm = ChatAnthropic(model="claude-3-5-haiku-20241022", temperature=0)
eval_result = audit_llm.invoke(prompt_query).content

audit_semantic_cache.store(
    prompt=prompt_query,
    response=eval_result,
    metadata={"node": "audit_verification"}
)

return {
    "audit_passed": eval_result == "PASSED",
    "audit_logs": [f"[Audit Node]: 신규 모델 검수 완료 ({eval_result})"]
}

Definimos o distance_threshold de forma rigorosa em 0.1 para evitar falsos positivos e chamamos o modelo leve Haiku apenas quando não houver correspondência no cache. Essa estrutura por si só pode reduzir os custos de tokens da API em até 40%, mantendo a qualidade geral da auditoria.

O que se faz necessário ao colocar um sistema de agentes em produção não são truques exóticos de prompt. É uma base sólida de engenharia backend — com isolamento de estado, controle de concorrência, retomada por checkpoints e tiering de modelos — que garante que o sistema não vá ruir.