LangGraph 멀티 에이전트가 프로덕션에서 터지는 이유와 코드 레벨 복구법

단일 프롬프트 체인을 넘어서 멀티 에이전트로 넘어간 백엔드 개발자가 흔히 겪는 착각이 있습니다. 프롬프트를 더 잘 쓰면 시스템이 안정될 거라는 생각입니다. 현장에서 터지는 문제 대부분은 프롬프트와 아무 상관이 없습니다. 상태 오염, 무한 루프, API Rate Limit, 디버깅 불가능한 비동기 트레이스 같은 시스템 구조 문제가 진짜 원인입니다.

LangGraph 기반 멀티 에이전트를 운영 환경에 올리려면 프롬프트 자원 집합이 아니라 백엔드 시스템을 다루듯 상태 격리, 동시성 제어, 트레이싱을 코드 레벨에서 잡아내야 합니다.

동시 실행 노드의 상태 오염과 무한 루프 차단

상태 기반 그래프에서 여러 노드가 하나의 공유 객체를 직접 수정하면 경합 상태가 일어납니다. 동시 실행되는 fan-out 패턴에서 별도 리듀서 없이 일반 필드를 덮어쓰면, 가장 늦게 끝난 노드의 결과만 남고 나머지는 날아갑니다.

이를 막으려면 상위 그래프 상태에는 명시적 리듀서를 붙이고, 하위 에이전트는 독립된 스키마를 갖는 서브그래프로 완전히 캡슐화해야 합니다.

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

상위 ParentState에서 병렬 쓰기가 발생하는 audit_logs 필드에 operator.add 리듀서를 지정했습니다. 하위 작업은 InternalAgentState라는 자체 상태를 사용하는 서브그래프로 격리한 뒤 래퍼 함수를 통해서만 결과를 주고받습니다. 데이터 오염이 차단되면서 무한 루프 디버깅 시간을 주당 5시간 이상 줄일 수 있습니다.

ReAct 피드백 루프가 종료 조건을 만족하지 못해 빙빙 도는 문제도 흔합니다. 상태 스키마에 카운터를 두고 조건부 엣지에서 이를 걸러내는 가드레일 루터가 필요합니다.

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

카운터(iterations)가 설정치(max_iterations)에 도달하면 즉시 수동 이관 노드(fallback_escalation)로 분기하도록 설정했습니다. 무한 순환으로 인한 토큰 낭비를 확실하게 끊어냅니다.

동시성 제어로 HTTP 429 에러 방지하기

하위 노드를 한 번에 병렬로 쏘아 올리면 OpenAI나 Anthropic API의 분당 토큰 제한(TPM)에 걸려 HTTP 429 에러가 터집니다. 백엔드가 흔들리는 순간 전체 트랜잭션이 마비됩니다.

asyncio.Semaphore로 동시 요청 수를 죄고 tenacity의 지수 백오프(Exponential Backoff)를 걸어야 API가 터지지 않습니다.

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

동시 호출을 최대 5개로 제한하고 실패 시 2초부터 10초까지 대기 시간을 두 배씩 늘리며 재시도합니다. 외부 API 호출 실패로 인한 시스템 중단을 완전히 막을 수 있습니다.

단일 루프 방식은 노드 하나만 터져도 전체 재추론에 14초 이상을 써야 하지만, 이처럼 노드를 격리하고 백오프를 걸어두면 실패 복구 시간을 10ms 수준으로 낮출 수 있습니다. 성공한 노드의 결과는 그대로 유지되므로 불필요한 토큰 재소비도 없습니다.

분산 트레이싱과 PostgresSaver 체크포인트 재개

비동기로 얽힌 에이전트는 콘솔 출력만으로 흐름을 잡을 수 없습니다. Langfuse 같은 OpenTelemetry 기반 관측 플랫폼을 붙이고, DB 연산이나 백엔드 처리 같은 비-LLM 로직까지 트레이싱 스팬에 넣어봐야 병목이 보입니다.

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)

@observe 데코레이터를 통해 vector search 같은 일반 함수도 트레이스에 집어넣습니다. CallbackHandler를 그래프 호출 시 넘겨주면 세션별로 전체 에이전트의 움직임과 토큰 소비량을 한눈에 확인할 수 있습니다.

실행 중간에 10번째 노드에서 에러가 났을 때 처음부터 다시 실행하면 돈과 시간이 동시에 날아갑니다. PostgresSaver 체크포인터를 쓰면 노드 실행 스냅샷이 DB에 그대로 저장됩니다.

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

get_state_history로 마지막 정상 상태를 확인한 뒤, update_state로 문제가 된 데이터를 고쳐 넣고 app.invoke(None, config)를 부르면 정확히 멈췄던 지점부터 다시 시작합니다.

모델 믹스와 시맨틱 캐시로 토큰 비용 40% 절감하기

모든 에이전트 노드에 GPT-4o를 때려 박는 건 예산 낭비입니다. 메인 플래닝에는 고성능 모델을 쓰고, 단순 분류나 검수 노드에는 Claude 3.5 Haiku 같은 경량 모델을 붙이는 티어링 기법이 기본입니다.

여기에 RedisVL 기반 시맨틱 캐시까지 걸어두면 동일하거나 유사한 검수 요청은 LLM을 호출하지도 않고 50ms 만에 반환합니다.

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

distance_threshold를 0.1로 엄격하게 잡아 오탐을 방지하고, 캐시에 없을 때만 경량 모델인 Haiku를 호출합니다. 이 구성만 맞춰도 전체 검수 품질을 유지하면서 API 토큰 비용을 최대 40%까지 내려버릴 수 있습니다.

에이전트 시스템을 프로덕션에 올릴 때 필요한 건 신기한 프롬프트 기교가 아닙니다. 상태 격리, 동시성 제어, 체크포인트 재개, 모델 티어링 같은 백엔드 기초 공사가 단단해야 시스템이 안 무너집니다.

LangGraph 멀티 에이전트가 프로덕션에서 터지는 이유와 코드 레벨 복구법

동시 실행 노드의 상태 오염과 무한 루프 차단

이를 막으려면 상위 그래프 상태에는 명시적 리듀서를 붙이고, 하위 에이전트는 독립된 스키마를 갖는 서브그래프로 완전히 캡슐화해야 합니다.

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

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

동시성 제어로 HTTP 429 에러 방지하기

asyncio.Semaphore로 동시 요청 수를 죄고 tenacity의 지수 백오프(Exponential Backoff)를 걸어야 API가 터지지 않습니다.

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

분산 트레이싱과 PostgresSaver 체크포인트 재개

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)

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

모델 믹스와 시맨틱 캐시로 토큰 비용 40% 절감하기

여기에 RedisVL 기반 시맨틱 캐시까지 걸어두면 동일하거나 유사한 검수 요청은 LLM을 호출하지도 않고 50ms 만에 반환합니다.

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

LangGraph 멀티 에이전트가 프로덕션에서 터지는 이유와 코드 레벨 복구법

Related Video

루프 엔지니어링은 가라, 이제 그래프 엔지니어링의 시대다

LangGraph 멀티 에이전트가 프로덕션에서 터지는 이유와 코드 레벨 복구법

동시 실행 노드의 상태 오염과 무한 루프 차단

동시성 제어로 HTTP 429 에러 방지하기

분산 트레이싱과 PostgresSaver 체크포인트 재개

모델 믹스와 시맨틱 캐시로 토큰 비용 40% 절감하기

Comments (0)

LangGraph 멀티 에이전트가 프로덕션에서 터지는 이유와 코드 레벨 복구법

동시 실행 노드의 상태 오염과 무한 루프 차단

동시성 제어로 HTTP 429 에러 방지하기

분산 트레이싱과 PostgresSaver 체크포인트 재개

모델 믹스와 시맨틱 캐시로 토큰 비용 40% 절감하기