Warum LangGraph-Multi-Agenten in der Produktion abstürzen und wie man sie auf Code-Ebene repariert

Backend-Entwickler, die den Schritt von einzelnen Prompt-Ketten zu Multi-Agenten-Systemen wagen, unterliegen häufig einem Irrtum: Sie glauben, dass ein besseres Prompting das System stabiler macht. Die meisten Probleme, die in der Praxis auftreten, haben jedoch überhaupt nichts mit Prompts zu tun. Die wahren Ursachen sind strukturelle Systemprobleme wie Zustandskontamination (State Pollution), Endlosschleifen, API-Rate-Limits und nicht-debuggbare asynchrone Traces.

Um LangGraph-basierte Multi-Agenten in einer Produktionsumgebung zu betreiben, muss man sie wie ein Backend-System behandeln – nicht wie eine bloße Sammlung von Prompt-Ressourcen. Zustandstrennung, Concurrency Control und Tracing müssen direkt auf Code-Ebene gelöst werden.

Zustandskontamination bei parallelen Knoten und das Verhindern von Endlosschleifen

In einem zustandsbasierten Graphen führt der direkte Zugriff mehrerer Knoten auf ein gemeinsam genutztes Objekt zu Race Conditions. Wenn Sie bei einem simultan ausgeführten Fan-Out-Muster ein normales Feld ohne expliziten Reducer überschreiben, bleiben nur die Ergebnisse des zuletzt fertiggestellten Knotens erhalten – der Rest geht verloren.

Um dies zu verhindern, müssen Sie dem übergeordneten Graphenzustand einen expliziten Reducer zuweisen und untergeordnete Agenten als Subgraphen mit einem unabhängigen Schema vollständig kapseln.

`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()
`

Im übergeordneten ParentState haben wir für das Feld audit_logs, in dem parallele Schreibzugriffe stattfinden, den Reducer operator.add definiert. Untergeordnete Aufgaben werden in einen Subgraphen mit eigenem Zustand (InternalAgentState) isoliert, und Daten werden ausschließlich über eine Wrapper-Funktion ausgetauscht. Durch das Verhindern von Datenkontamination lässt sich die Debugging-Zeit für Endlosschleifen um mehr als 5 Stunden pro Woche reduzieren.

Ein weiteres häufiges Problem ist das Feststecken in ReAct-Feedbackschleifen, wenn die Abbruchbedingung nicht erreicht wird. Hierfür wird ein Guardrail-Router benötigt, der einen Zähler im Zustandsschema führt und diesen an bedingten Kanten (Conditional Edges) prüft.

`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()
`

Sobald der Zähler (iterations) den festgelegten Schwellenwert (max_iterations) erreicht, verzweigt das System sofort zum manuellen Eskalationsknoten (fallback_escalation). Dies stoppt die unnötige Token-Verschwendung durch Endlosschleifen zuverlässig.

HTTP-429-Fehler durch Concurrency Control vermeiden

Wenn untergeordnete Knoten massenhaft und gleichzeitig parallel ausgeführt werden, stoßen Sie schnell an das Token-pro-Minute-Limit (TPM) von APIs wie OpenAI oder Anthropic, was zu HTTP-429-Fehlern führt. Sobald das Backend ins Wanken gerät, wird die gesamte Transaktion gelähmt.

Mit asyncio.Semaphore müssen Sie die Anzahl paralleler Anfragen begrenzen und mit tenacity einen exponentiellen Backoff (Exponential Backoff) einrichten, um API-Abstürze zu verhindern.

`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]}
`

In diesem Beispiel werden gleichzeitige Aufrufe auf maximal 5 beschränkt. Im Fehlerfall verdoppelt sich die Wartezeit bei jedem Versuch schrittweise von 2 auf bis zu 10 Sekunden. Dadurch lassen sich Systemausfälle durch fehlgeschlagene externe API-Aufrufe komplett verhindern.

Bei einem monolithischen Schleifenansatz führt der Ausfall eines einzelnen Knotens dazu, dass die gesamte Re-Inferenz mehr als 14 Sekunden dauert. Wenn Sie die Knoten jedoch wie beschrieben isolieren und mit Backoff absichern, lässt sich die Zeit zur Fehlerbehebung auf etwa 10 ms senken. Da die Ergebnisse erfolgreich ausgeführter Knoten erhalten bleiben, fällt auch kein unnötiger erneuter Token-Verbrauch an.

Verteiltes Tracing und Neustart per PostgresSaver-Checkpoint

Bei asynchron miteinander verwobenen Agenten reicht eine bloße Konsolenausgabe nicht aus, um den Ablauf nachzuvollziehen. Sie müssen eine OpenTelemetry-basierte Observability-Plattform wie Langfuse anbinden und auch Nicht-LLM-Logik wie DB-Operationen oder Backend-Prozesse in Spans erfassen, um Engpässe sichtbar zu machen.

`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)
`

Über den Decorator @observe werden auch reguläre Funktionen wie die Vektorsuche in das Tracing integriert. Wenn Sie den CallbackHandler beim Aufruf des Graphen übergeben, können Sie das Verhalten des gesamten Agenten sowie den Token-Verbrauch pro Sitzung auf einen Blick überwachen.

Tritt während der Ausführung beim 10. Knoten ein Fehler auf, ist ein vollständiger Neustart von vorne sowohl zeit- als auch kostenintensiv. Mit dem PostgresSaver-Checkpointer werden Snapshots der Knotenausführung direkt in der Datenbank gespeichert.

`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

Nachdem Sie den letzten fehlerfreien Zustand mit get_state_history überprüft haben, korrigieren Sie die fehlerhaften Daten mit update_state und rufen app.invoke(None, config) auf. Die Ausführung wird exakt an der Stelle fortgesetzt, an der sie unterbrochen wurde.

Token-Kosten um 40% senken durch Model-Mix und Semantic Caching

GPT-4o für jeden einzelnen Agentenknoten einzusetzen, ist reine Budgetverschwendung. Standard sollte ein Tiering-Ansatz sein: Verwenden Sie ein Hochleistungsmodell für das primäre Planning und binden Sie Leichtgewicht-Modelle wie Claude 3.5 Haiku für einfache Klassifizierungs- oder Überprüfungsknoten ein.

Wenn Sie zusätzlich ein Semantic Caching auf Basis von RedisVL schalten, werden identische oder ähnliche Überprüfungsanfragen in nur 50 ms beantwortet – ohne überhaupt ein LLM aufzurufen.

`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})"]
}

Mit einem strikt eingestellten distance_threshold von 0.1 werden False Positives vermieden. Nur wenn der Cache keinen Treffer liefert, wird das leichtgewichtige Haiku-Modell aufgerufen. Allein durch diese Konfiguration lassen sich die API-Token-Kosten um bis zu 40% senken, während die Qualität der Überprüfung aufrechterhalten bleibt.

Um ein Agentensystem erfolgreich in die Produktion zu überführen, bedarf es keiner ausgefallenen Prompt-Tricks. Es ist ein solides Backend-Fundament – bestehend aus Zustandstrennung, Concurrency Control, Checkpoint-Resumption und Modell-Tiering –, das verhindert, dass das System zusammenbricht.