Por qué los multiagentes de LangGraph fallan en producción y cómo solucionarlo a nivel de código
makedream٢٦ يوليو ٢٠٢٦
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
Existe un error común entre los desarrolladores backend que pasan de cadenas de prompts simples a sistemas multiagente: pensar que escribir mejores prompts hará que el sistema sea más estable. La mayoría de los problemas en producción no tienen nada que ver con los prompts. Las causas reales son problemas de arquitectura de sistemas, como la contaminación del estado, los bucles infinitos, los límites de tasa (Rate Limit) de la API y las trazas asíncronas imposibles de depurar.
Para llevar un sistema multiagente basado en LangGraph a un entorno de producción, es necesario gestionar la aislación de estado, el control de concurrencia y la trazabilidad a nivel de código, exactamente igual que se haría con cualquier sistema backend, en lugar de tratarlo como un mero conjunto de recursos de prompts.
En un grafo basado en estado, si varios nodos modifican un mismo objeto compartido de forma directa, se producen condiciones de carrera (race conditions). En un patrón de abanico de salida (fan-out) con ejecución concurrente, si se sobrescribe un campo normal sin usar un reductor (reducer) específico, solo se conservará el resultado del último nodo en finalizar y el resto se perderá.
Para evitar esto, se debe asignar un reductor explícito al estado del grafo superior y encapsular por completo los agentes inferiores como subgrafos con sus propios esquemas independientes.
`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()
`
En el estado superior ParentState, se asignó el reductor operator.add al campo audit_logs, donde ocurren escrituras en paralelo. Las tareas secundarias se aislaron en un subgrafo que utiliza su propio estado InternalAgentState, intercambiando resultados únicamente a través de una función wrapper. Al bloquear la contaminación de datos, es posible reducir el tiempo de depuración de bucles infinitos en más de 5 horas a la semana.
También es frecuente el problema en el que el bucle de retroalimentación de ReAct no logra cumplir la condición de salida y gira indefinidamente. Se requiere un enrutador con barreras de seguridad (guardrails) que mantenga un contador en el esquema de estado y lo filtre en los bordes condicionales.
`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()
`
Se ha configurado para que, en cuanto el contador (iterations) alcance el valor establecido (max_iterations), se desvíe de inmediato hacia el nodo de derivación manual (fallback_escalation). Esto corta de forma drástica el desperdicio de tokens causado por iteraciones infinitas.
Si se ejecutan múltiples nodos secundarios en paralelo simultáneamente, se alcanzará el límite de tokens por minuto (TPM) de las API de OpenAI o Anthropic, desencadenando errores HTTP 429. En el momento en que el backend se desestabiliza, la transacción completa se paraliza.
Para evitar que la API falle, es necesario limitar el número de solicitudes concurrentes mediante asyncio.Semaphore y aplicar un retroceso exponencial (exponential backoff) con tenacity.
`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]}
`
Se limita el número de llamadas simultáneas a un máximo de 5 y, en caso de fallo, se reintenta duplicando el tiempo de espera secuencialmente desde 2 hasta 10 segundos. Esto previene por completo la interrupción del sistema debido a fallos en las llamadas a API externas.
En una arquitectura de bucle único, el fallo de un solo nodo obliga a gastar más de 14 segundos en reejecutar toda la inferencia; sin embargo, al aislar los nodos y aplicar un retroceso exponencial, el tiempo de recuperación ante fallos se reduce a tan solo 10 ms. Además, los resultados de los nodos ejecutados con éxito se conservan, evitando el consumo innecesario de tokens.
No es posible rastrear el flujo de un sistema de agentes asíncronos complejos únicamente mediante impresiones en consola. Es necesario integrar una plataforma de observabilidad basada en OpenTelemetry, como Langfuse, e incluir dentro de los spans de trazabilidad la lógica ajena al LLM, como las operaciones de base de datos o el procesamiento en el backend, para identificar los cuellos de botella.
`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)
`
A través del decorador @observe, se incluyen funciones comunes como la búsqueda vectorial dentro de la traza. Al pasar el CallbackHandler durante la invocación del grafo, se puede visualizar con claridad el comportamiento de todos los agentes y el consumo de tokens por sesión.
Si ocurre un error en el décimo nodo durante la ejecución, volver a ejecutar todo desde el principio desperdicia tiempo y dinero simultáneamente. Al utilizar el guardador de puntos de comprobación (checkpointer) PostgresSaver, las capturas de estado (snapshots) de la ejecución de cada nodo se almacenan directamente en la base de datos.
`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
`
Tras verificar el último estado válido mediante get_state_history, se corrigen los datos con problemas usando update_state y se invoca app.invoke(None, config) para reanudar la ejecución exactamente desde el punto en que se detuvo.
Asignar GPT-4o a todos los nodos de los agentes representa un desperdicio de presupuesto. La técnica de estratificación (tiering), que consiste en emplear modelos de alto rendimiento para la planificación principal y conectar modelos ligeros como Claude 3.5 Haiku a nodos de clasificación simple o auditoría, resulta fundamental.
Si además se añade una capa de caché semántica basada en RedisVL, las solicitudes de auditoría idénticas o similares se responderán en tan solo 50 ms sin necesidad de realizar llamadas al 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})"]
}
`
Al establecer un umbral de distancia estricto de 0.1 (distance_threshold) se evitan falsos positivos, llamando al modelo ligero Haiku únicamente cuando no hay coincidencias en la caché. Solo con esta configuración es posible reducir los costos de tokens de la API hasta en un 40% manteniendo la calidad general de la auditoría.
Lo que realmente se necesita para desplegar un sistema de agentes en producción no son trucos o técnicas deslumbrantes de ingeniería de prompts. Para evitar que el sistema se colapse, se requiere una base sólida de arquitectura backend que incluya aislación de estado, control de concurrencia, reanudación desde puntos de comprobación y estratificación de modelos.