Alasan Mengapa Multi-Agent LangGraph Hancur di Produksi dan Cara Memulihkannya di Level Kode

Ada kesalahpahaman umum di antara para pengembang backend yang bertransisi dari chain prompt tunggal ke sistem multi-agent. Mereka berpikir bahwa menulis prompt yang lebih baik akan membuat sistem menjadi stabil. Namun, sebagian besar masalah yang terjadi di lapangan sama sekali tidak ada hubungannya dengan prompt. Masalah struktur sistem seperti kontaminasi status (state contamination), loop tak terbatas (infinite loop), API Rate Limit, serta asynchronous trace yang mustahil di-debug adalah penyebab utamanya.

Untuk menjalankan multi-agent berbasis LangGraph di lingkungan produksi, Anda harus menangani isolasi status, kontrol konkurensi, dan tracing di level kode layaknya mengelola sistem backend, bukan sekadar memperlakukannya sebagai kumpulan sumber daya prompt.

Mencegah Kontaminasi Status Node Eksekusi Konkuren dan Loop Tak Terbatas

Dalam grafik berbasis status, jika beberapa node secara langsung mengubah satu objek bersama, maka race condition akan terjadi. Dalam pola fan-out yang berjalan secara konkuren, jika Anda menimpa field biasa tanpa reducer terpisah, hanya hasil dari node yang paling terakhir selesailah yang tersisa, sedangkan sisanya akan hilang.

Untuk mencegah hal ini, Anda harus menetapkan reducer eksplisit pada status parent graph, dan mengenkapsulasi sub-agent sepenuhnya ke dalam subgraph yang memiliki skema independen.

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

Pada ParentState tingkat atas, reducer operator.add ditentukan untuk field audit_logs di mana penulisan paralel terjadi. Sub-task diisolasi ke dalam subgraph yang menggunakan statusnya sendiri yang disebut InternalAgentState, dan hasil hanya dipertukarkan melalui fungsi wrapper. Dengan memblokir kontaminasi data, waktu debugging loop tak terbatas dapat dipangkas lebih dari 5 jam per minggu.

Masalah lain yang sering terjadi adalah ReAct feedback loop yang terus berputar karena tidak memenuhi kondisi berhenti. Anda memerlukan guardrail router yang menempatkan counter pada skema status dan menyaringnya pada conditional edge.

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

Sistem dikonfigurasi untuk segera bercabang ke node eskalasi manual (fallback_escalation) ketika counter (iterations) mencapai nilai batas (max_iterations). Hal ini secara efektif menghentikan pemborosan token akibat perulangan tak terbatas.

Mencegah Error HTTP 429 dengan Kontrol Konkurensi

Memicu sub-node secara paralel sekaligus dapat menyebabkan error HTTP 429 karena melampaui batas token per menit (TPM) pada API OpenAI atau Anthropic. Begitu backend mengalami kegagalan, seluruh transaksi akan lumpuh.

Anda harus membatasi jumlah permintaan konkuren menggunakan asyncio.Semaphore dan menerapkan Exponential Backoff dari tenacity agar API tidak tumbang.

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

Panggilan konkuren dibatasi maksimal 5, dan jika terjadi kegagalan, sistem akan mencoba lagi dengan mengandakan waktu tunggu mulai dari 2 detik hingga 10 detik. Hal ini dapat sepenuhnya mencegah penghentian sistem akibat kegagalan pemanggilan API eksternal.

Dalam pendekatan single-loop, jika satu node saja mengalami error, Anda harus menghabiskan lebih dari 14 detik untuk melakukan inferensi ulang secara keseluruhan. Namun, dengan mengisolasi node dan menerapkan backoff seperti ini, waktu pemulihan dari kegagalan dapat dipangkas hingga sekitar 10ms. Karena hasil dari node yang berhasil tetap dipertahankan, tidak ada pemborosan konsumsi token yang tidak perlu.

Distributed Tracing dan Melanjutkan Eksekusi via PostgresSaver Checkpoint

Agent yang terhubung secara asinkron tidak dapat dilacak jalannya hanya dengan keluaran konsol. Anda perlu menghubungkan platform observabilitas berbasis OpenTelemetry seperti Langfuse, dan memasukkan logika non-LLM seperti operasi DB atau pemrosesan backend ke dalam span tracing untuk mengidentifikasi bottleneck.

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

Fungsi biasa seperti vector search juga dimasukkan ke dalam trace melalui dekorator @observe. Dengan meneruskan CallbackHandler saat memanggil graph, Anda dapat melihat seluruh pergerakan agent dan konsumsi token per sesi secara sekilas.

Jika terjadi error pada node ke-10 di tengah-tengah eksekusi, menjalankan ulang dari awal akan membuang uang dan waktu secara bersamaan. Dengan menggunakan checkpointer PostgresSaver, snapshot eksekusi node akan disimpan secara utuh di DB.

`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

Setelah memeriksa status normal terakhir dengan get_state_history, Anda dapat memperbarui data yang bermasalah menggunakan update_state lalu memanggil app.invoke(None, config) untuk melanjutkan eksekusi tepat dari titik di mana sistem terhenti.

Menghemat Biaya Token Hingga 40% dengan Model Mix dan Semantic Cache

Menggunakan GPT-4o di setiap node agent adalah pemborosan anggaran. Teknik peringkatan (tiering) adalah standar utama: gunakan model berkinerja tinggi untuk perencanaan utama, dan gunakan model ringan seperti Claude 3.5 Haiku untuk node klasifikasi sederhana atau verifikasi.

Jika ditambah dengan semantic cache berbasis RedisVL, permintaan verifikasi yang identik atau serupa bahkan tidak akan memanggil LLM dan akan memberikan respons hanya dalam 50ms.

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

Dengan mengatur distance_threshold secara ketat pada 0.1 untuk mencegah false positive, model ringan Haiku hanya dipanggil saat data tidak ada di cache. Hanya dengan menerapkan konfigurasi ini, Anda dapat memangkas biaya token API hingga 40% sambil tetap mempertahankan kualitas verifikasi secara keseluruhan.

Hal yang dibutuhkan saat membawa sistem agent ke lingkungan produksi bukanlah teknik prompt yang unik. Fondasi backend yang kokoh—seperti isolasi status, kontrol konkurensi, pemulihan checkpoint, dan tiering model—adalah hal yang menjaga agar sistem tidak runtuh.