Pourquoi les multi-agents LangGraph plantent en production et comment les réparer au niveau du code
makedream26. Juli 2026
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
Il existe une idée reçue très courante chez les développeurs backend qui passent d'une simple chaîne de prompts à une architecture multi-agents : penser qu'en rédigeant de meilleurs prompts, le système deviendra plus stable. En réalité, la plupart des problèmes rencontrés sur le terrain n'ont absolument rien à voir avec les prompts. La vraie cause réside dans des problèmes d'architecture système, tels que la contamination d'état, les boucles infinies, les limites de débit API (Rate Limits) et les traces asynchrones impossibles à déboguer.
Pour passer des multi-agents basés sur LangGraph en environnement de production, il ne faut pas les traiter comme un simple ensemble de ressources de prompts, mais contrôler l'isolation d'état, la gestion de la concurrence et le traçage au niveau du code, exactement comme on le ferait pour un système backend.
Dans un graphe orienté état, lorsque plusieurs nœuds modifient directement un même objet partagé, des conditions de concurrence (race conditions) surviennent. Dans un schéma d'exécution parallèle de type fan-out, si vous écrasez un champ classique sans utiliser de réducteur (reducer) dédié, seul le résultat du nœud qui s'est terminé en dernier sera conservé, et les autres seront perdus.
Pour éviter cela, il faut rattacher un réducteur explicite à l'état du graphe parent, et encapsuler entièrement l'agent enfant sous forme de sous-graphe possédant son propre schéma indépendant.
`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()
`
Dans le ParentState parent, nous avons spécifié le réducteur operator.add pour le champ audit_logs soumis à des écritures parallèles. Les sous-tâches sont isolées dans un sous-graphe utilisant son propre état InternalAgentState, et les résultats ne sont échangés qu'à travers une fonction wrapper. En bloquant la contamination des données, vous pouvez réduire le temps de débogage des boucles infinies de plus de 5 heures par semaine.
Les boucles de rétroaction ReAct qui tournent en rond faute de satisfaire aux conditions de sortie sont également très fréquentes. Il est indispensable de placer un compteur dans le schéma d'état et d'installer un routeur garde-fou (guardrail router) pour filtrer cela au niveau des arêtes conditionnelles.
`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()
`
Dès que le compteur (iterations) atteint la valeur configurée (max_iterations), le flux se dirige immédiatement vers le nœud d'escalade manuelle (fallback_escalation). Cela permet de couper net le gaspillage de jetons causé par des boucles infinies.
Si vous lancez plusieurs nœuds enfants en parallèle simultanément, vous allez vous heurter à la limite de jetons par minute (TPM) des API OpenAI ou Anthropic, ce qui déclenchera des erreurs HTTP 429. Dès que le backend flanche, c'est l'ensemble de la transaction qui se retrouve paralysé.
Pour éviter l'effondrement de l'API, il faut restreindre le nombre de requêtes simultanées à l'aide d'un asyncio.Semaphore et appliquer une stratégie de backoff exponentiel avec 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]}
`
En limitant les appels simultanés à 5 au maximum et en doublant le temps d'attente en cas d'échec (de 2 à 10 secondes), vous tentez de nouveau la requête. Cela permet de prévenir totalement les interruptions système dues à des échecs d'appels d'API externes.
Avec une approche монолитique à boucle unique, le crash d'un seul nœud oblige à relancer tout le processus d'inférence pendant plus de 14 secondes. En revanche, en isolant ainsi les nœuds et en appliquant un backoff, vous réduisez le temps de récupération après échec à environ 10 ms. Les résultats des nœuds ayant réussi étant préservés, il n'y a pas non plus de reconsommation inutile de jetons.
Quand des agents sont imbriqués de manière asynchrone, la simple sortie console ne suffit plus pour comprendre le flux. Il faut intégrer une plateforme d'observabilité basée sur OpenTelemetry comme Langfuse, et inclure dans les durées de traçage (spans) la logique non-LLM comme les opérations de base de données ou le traitement backend pour identifier les goulots d'étranglement.
`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)
`
Grâce au décorateur @observe, même les fonctions classiques comme la recherche vectorielle sont incluses dans le trace. En passant le CallbackHandler lors de l'appel du graphe, vous pouvez visualiser en un coup d'œil l'activité globale des agents et la consommation de jetons par session.
Si une erreur survient au 10ème nœud en cours d'exécution, tout relancer depuis le début ferait perdre à la fois du temps et de l'argent. L'utilisation du checkpointer PostgresSaver permet de sauvegarder l'instantané (snapshot) d'exécution des nœuds directement en base de données.
`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
`
Après avoir vérifié le dernier état valide avec get_state_history, il suffit de corriger les données problématiques avec update_state et d'appeler app.invoke(None, config) pour reprendre l'exécution exactement là où elle s'était arrêtée.
Injecter GPT-4o dans tous les nœuds d'agents est un gaspillage de budget. La technique de basique de tiering consiste à utiliser un modèle haute performance pour la planification principale, tout en rattachant des modèles plus légers comme Claude 3.5 Haiku aux nœuds de classification simple ou de vérification.
En y ajoutant un cache sémantique basé sur RedisVL, les requêtes de vérification identiques ou similaires seront renvoyées en seulement 50 ms sans même solliciter le 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})"]
}
`
En réglant strictement le distance_threshold à 0.1 pour éviter les faux positifs, le modèle léger Haiku n'est appelé que lorsque la réponse n'est pas en cache. Rien qu'avec cette configuration, vous pouvez réduire les coûts de jetons API jusqu'à 40 % tout en maintenant la qualité globale de révision.
Ce dont vous avez besoin pour passer un système d'agents en production, ce ne sont pas de simples astuces de prompting originales. C'est une fondation backend solide — isolation d'état, contrôle de concurrence, reprise par points de contrôle, tiering de modèles — qui garantira que votre système ne s'effondre pas.