LangGraph는 Architecture가 아니다: Runtime의 정확한 위치

LangGraph로 Agent를 만들기 시작하면 어느 순간 모든 것이 graph 안으로 빨려 들어간다. 사용자의 질문도 state에 넣고, 검색 결과도 state에 넣고, 진행률도 state에 넣고, 최종 보고서도 state에 넣는다. thread_id가 있으니 session도 해결된 듯하고, checkpointer가 있으니 데이터베이스도 해결된 듯하다. graph.astream()이 값을 내보내니 streaming API까지 완성된 것처럼 보인다.

로컬 notebook에서는 이 그림이 꽤 오래 버틴다.

User Request
    ↓
LangGraph
    ├─ search node
    ├─ analyze node
    └─ write node
    ↓
Final Answer

운영 환경에 올리는 순간 graph 바깥의 질문이 생긴다.

• HTTP 요청이 끊기면 graph 실행도 취소할 것인가?
• 실행 대기 중인 run은 누가 소유하는가?
• 브라우저에 보여줄 진행률과 graph 내부 문서를 같은 state에 둘 것인가?
• Worker가 죽으면 queue 메시지와 checkpoint를 어떻게 함께 복구할 것인가?
• 사용자가 다른 사람의 thread_id를 넣으면 누가 막는가?
• Framework event 형식이 바뀌면 SSE client도 함께 바꿀 것인가?

이 질문은 LangGraph의 부족함을 지적하는 것이 아니다. Runtime과 아키텍처가 서로 다른 범위의 문제를 푼다는 뜻이다.

Runtime은 실행을 진행시키고, 아키텍처는 실행을 둘러싼 계약을 만든다

이 시리즈에서 사용할 위치는 다음과 같다.

Browser
   │ POST /runs, GET /events
   ▼
API ── authentication / authorization / admission
   │
   ▼
Queue ── pending work / delivery / redelivery
   │
   ▼
Worker ── attempt lifecycle / timeout / cancellation
   │
   ▼
LangGraph ── node execution / transition / checkpoint / runtime stream
   │
   ├─ updates ──> Redis UI state
   ├─ custom  ──> Redis event stream ──> SSE
   └─ result  ──> PostgreSQL business record

LangGraph는 worker 안에서 실행된다. 이 배치는 graph가 중요하지 않다는 뜻이 아니다. 오히려 graph가 무엇을 확실히 책임지는지 분명하게 만든다.

Layer	주된 책임
API	사용자 인증, run 생성 권한, 입력 검증, 202 Accepted
Queue	아직 처리되지 않은 작업의 영속적 전달
Worker	활성 실행 시도, heartbeat, 취소, timeout, 확인 처리
LangGraph	node 일정 관리, state 갱신, 조건부 전환, checkpoint/재개
Redis	빠른 UI snapshot과 replay 가능한 진행 event
PostgreSQL	run의 기준 기록과 최종 결과
SSE gateway	cursor 이후 event를 브라우저에 전달

제품에 따라 이 경계는 겹칠 수 있다. LangGraph 기반 서버 제품이 영속성이나 run API까지 제공할 수도 있다. 그래도 어느 구성 요소가 어떤 보장을 제공하는지를 묻는 일은 사라지지 않는다. Library를 직접 worker에 넣는 구조라면 위 책임은 애플리케이션이 명시적으로 조립해야 한다.

먼저 graph state를 작게 정의한다

예제로 NVIDIA의 최근 실적과 valuation을 분석하는 graph를 생각해 보자.

from operator import add
from typing import Annotated, TypedDict


class AnalysisState(TypedDict):
    query: str
    documents: list[dict]
    analysis: dict | None
    report: str | None
    status: str
    warnings: Annotated[list[str], add]

여기서 state는 graph node들이 다음 계산을 하기 위해 공유하는 runtime 작업 집합이다.

• query: 검색과 분석이 읽는다.
• documents: 검색 결과를 분석 node가 읽는다.
• analysis: 보고서 node가 읽는다.
• status: graph 안의 현재 단계를 표현한다.
• warnings: 여러 node가 발견한 경고를 누적한다.

이 state를 곧바로 브라우저 응답 schema나 데이터베이스 table로 생각하면 안 된다. documents에는 수백 KB의 원문과 제공자 metadata가 들어갈 수 있고, analysis에는 내부 점수와 prompt용 중간 결과가 포함될 수 있다. 브라우저에는 대부분 필요 없으며, tenant가 볼 수 없는 내부 자료가 섞일 수도 있다.

Graph state는 node 실행에 필요한 모양으로 설계한다. UI state는 사용자가 현재 화면을 복원하는 데 필요한 모양으로, 업무 기록은 감사와 제품 기능에 필요한 모양으로 설계한다.

Node는 전체 state가 아니라 update를 반환한다

검색 node는 전체 AnalysisState를 복사해 반환할 필요가 없다.

async def search_node(state: AnalysisState) -> dict:
    documents = await search_provider.search(state["query"])

    return {
        "documents": documents,
        "status": "analyzing",
    }

Node 실행 전 state가 다음과 같다고 하자.

{
    "query": "NVDA margin and valuation analysis",
    "documents": [],
    "analysis": None,
    "report": None,
    "status": "searching",
    "warnings": [],
}

검색 node는 documents와 status만 바꾼다. query, analysis, report, warnings는 기존 값을 유지한다. 이 부분 갱신이 LangGraph 내부 state에 적용될 때 각 key의 reducer가 사용된다.

공식 Graph API 문서의 규칙은 명확하다. Reducer를 지정하지 않은 key는 새 값으로 교체된다. Annotated[list[str], add]처럼 reducer를 지정한 key는 기존 값과 갱신을 reducer로 결합한다. (Graph API: Reducers)

Node output                    Graph state after reducer
───────────────────────────    ─────────────────────────────
documents = [doc1, doc2]   →   documents 교체
status = "analyzing"       →   status 교체
warnings = ["stale data"]  →   기존 warnings 뒤에 누적

Reducer는 단순한 typing 장식이 아니다. 병렬 분기 두 개가 같은 key를 갱신할 때 결합 규칙을 결정하는 runtime 의미 체계다. 반대로 add reducer가 붙은 list를 비우려고 []를 반환해도 기존 값이 사라지지 않는다. 빈 list를 더한 결과는 그대로이기 때문이다. 누적 field의 초기화 의미를 따로 설계해야 하는 이유다.

그러나 여기서 중요한 경계가 하나 있다.

LangGraph reducer는 graph state 갱신을 합치는 규칙이지, Redis와 PostgreSQL의 동시성 제어 규칙이 아니다.

Reducer가 있다고 Redis 쓰기가 원자적으로 바뀌거나, 두 worker가 같은 run을 실행하는 문제가 해결되거나, 데이터베이스 transaction이 만들어지는 것은 아니다.

updates는 state 전체가 아니라 node가 바꾼 부분을 보여준다

Graph를 구성한 뒤 graph.astream()으로 runtime output을 받을 수 있다.

async for part in graph.astream(
    {
        "query": request.query,
        "documents": [],
        "analysis": None,
        "report": None,
        "status": "searching",
        "warnings": [],
    },
    config={"configurable": {"thread_id": run_id}},
    stream_mode=["updates", "custom"],
    version="v2",
):
    event_type = part["type"]
    namespace = part["ns"]
    data = part["data"]

현재 LangGraph의 stream-mode 문서에서 version="v2"는 mode 수나 subgraph 사용 여부와 무관하게 {type, ns, data} 형태의 통일된 StreamPart를 제공한다. updates는 단계마다 node가 반환한 state 갱신을, custom은 node 내부에서 보낸 사용자 정의 데이터를 전달한다. (LangGraph Streaming)

검색 node가 끝난 뒤 updates part의 data는 대략 다음과 같다.

{
    "search_node": {
        "documents": [
            {"id": "10-k", "text": "..."},
            {"id": "earnings-call", "text": "..."},
        ],
        "status": "analyzing",
    }
}

이것은 완성된 graph state가 아니다. 어느 node가 어떤 부분을 갱신했는지를 나타낸다. 전체 state가 필요하면 values mode가 있지만, worker가 UI 투영값을 만들기 위해 매 단계마다 큰 문서와 분석 전체를 받을 이유는 대개 없다.

또 하나의 보안상 함의가 있다. Graph의 입력/출력 schema에서 비공개 채널처럼 취급한 field라도 streaming에는 노출될 수 있다. 따라서 runtime stream을 그대로 외부 client에 proxy하지 말고, 애플리케이션이 허용한 field만 별도 계약으로 변환해야 한다.

custom은 state가 아니라 실행 중 발생한 event에 가깝다

updates는 node가 끝나 state 갱신을 반환할 때 나온다. 그런데 문서 100개를 분석하는 node 하나가 40초 걸린다면, node가 끝날 때까지 아무 갱신도 볼 수 없다.

이때 custom stream을 사용해 node 내부 진행 상황을 보낼 수 있다.

from langgraph.config import get_stream_writer


async def analyze_documents(state: AnalysisState) -> dict:
    writer = get_stream_writer()
    results = []
    total = len(state["documents"])

    for index, document in enumerate(state["documents"]):
        result = await analyst.analyze(document)
        results.append(result)

        writer({
            "kind": "document_analyzed",
            "completed": index + 1,
            "total": total,
            "document_id": document["id"],
        })

    return {
        "analysis": aggregate(results),
        "status": "writing",
    }

document_analyzed는 graph의 영속적 작업 state를 바꾸기 위한 값이라기보다, 실행 도중 일어난 사실을 소비자에게 알리는 event다.

custom event                            final node update
────────────────────────────────────    ──────────────────────────
document 1 analyzed                     analysis = aggregate(...)
document 2 analyzed                     status = "writing"
document 3 analyzed
...

모든 진행 event를 graph state list에 계속 추가하면 checkpoint가 커지고 node 입력에도 불필요한 이력이 반복해서 들어간다. 반대로 재개 후 다음 계산에 반드시 필요한 정보라면 일시적 event만으로 두면 안 된다. 판단 기준은 “UI에 보이는가”가 아니라 다음 node가 정확히 실행되기 위해 필요한가다.

세 개의 state는 같을 필요가 없다

이 글의 핵심 구분은 다음 한 줄이다.

LangGraph State  ≠  Redis UI State  ≠  Browser UI State

각 state의 소비자와 수명이 다르다.

State	주요 소비자	포함할 것	피할 것
LangGraph state	Node와 runtime	다음 계산에 필요한 작업 데이터	UI 표현 전용 텍스트, 무한 event 이력
Redis UI state	API/SSE와 운영 화면	status, 단계, 진행률, 종료 요약	대형 문서, 비공개 prompt, 전체 보고서
브라우저 state	사용자 화면	표시 가능한 진행률과 결과 링크	내부 node payload, 권한 없는 metadata

예를 들어 graph 갱신에는 문서 원문이 포함되어도 Redis snapshot에는 단계만 반영할 수 있다.

PUBLIC_STATUS = {
    "searching": "자료를 찾는 중",
    "analyzing": "자료를 분석하는 중",
    "writing": "보고서를 작성하는 중",
    "completed": "완료",
}


async def project_update(redis, run_id: str, update: dict) -> None:
    for node_name, changed in update.items():
        status = changed.get("status")
        if status is None:
            continue

        await redis.hset(
            f"run:{run_id}:state",
            mapping={
                "status": status,
                "label": PUBLIC_STATUS[status],
                "last_node": node_name,
            },
        )

changed["documents"]를 Redis에 복제하지 않는 것이 의도다. Redis는 graph 메모리의 복제품이 아니라 UI와 운영을 위한 투영값이다.

브라우저 state는 한 번 더 좁아질 수 있다. 같은 Redis snapshot을 읽더라도 tenant 정책과 사용자 역할에 따라 비용이나 문서 이름을 숨길 수 있다. 이 변환은 framework state 병합이 아니라 애플리케이션 권한 확인의 영역이다.

Runtime event를 product event로 번역한다

Framework가 내보내는 원본 event를 그대로 SSE 계약으로 공개하면 client가 runtime 구현에 결합된다.

나쁜 결합
LangGraph node name / StreamPart payload
                ↓ 그대로 노출
Browser가 framework 내부 구조를 해석

권장 경계
LangGraph runtime output
                ↓ application mapper
Versioned product event
                ↓ Redis Stream / SSE
Browser가 안정된 contract를 해석

애플리케이션 mapper는 허용 목록, 검증과 version 관리를 담당한다.

def to_product_event(custom: dict) -> dict | None:
    if custom.get("kind") != "document_analyzed":
        return None

    completed = int(custom["completed"])
    total = int(custom["total"])

    if total <= 0 or not 0 <= completed <= total:
        raise ValueError("invalid progress payload")

    return {
        "type": "analysis_progressed",
        "version": 1,
        "payload": {
            "completed": completed,
            "total": total,
        },
    }

Graph node 이름을 analyze_documents_v2로 바꾸거나 내부 payload에 제공자 metadata를 추가해도 공개 event는 유지할 수 있다. 반대로 제품 UI 요구가 바뀌면 mapper와 event version을 의도적으로 바꾼다.

이 경계는 데이터 유출도 줄인다. custom writer에 실수로 query, prompt 또는 문서 텍스트가 들어와도 허용 목록에 없는 field는 Redis Stream과 브라우저로 넘어가지 않는다.

Worker가 runtime 출력을 애플리케이션 저장소에 반영한다

이제 worker의 역할을 한데 모을 수 있다.

import json


async def execute_run(run, graph, redis, repository) -> None:
    await repository.mark_running(run.id, run.attempt_id)

    try:
        async for part in graph.astream(
            initial_graph_state(run),
            config={"configurable": {"thread_id": run.checkpoint_key}},
            stream_mode=["updates", "custom"],
            version="v2",
        ):
            if part["type"] == "updates":
                await project_update(redis, run.id, part["data"])

            elif part["type"] == "custom":
                event = to_product_event(part["data"])
                if event is not None:
                    await redis.xadd(
                        f"run:{run.id}:events",
                        {"payload": json.dumps(event)},
                    )

        final_state = await graph.aget_state(
            {"configurable": {"thread_id": run.checkpoint_key}}
        )
        report = final_state.values["report"]

        await repository.complete_run(run.id, report)
        await redis.xadd(
            f"run:{run.id}:events",
            {"payload": json.dumps({"type": "run_succeeded", "version": 1})},
        )

    except Exception as error:
        await repository.fail_attempt(run.id, run.attempt_id, repr(error))
        raise

이 코드는 개념 예시다. 운영 환경에서는 fencing token, lease, 취소, 기한, 재시도 분류와 멱등성이 더 필요하다. 재개 후 Tool node의 부수 효과가 반복되지 않게 하려면 Agent Tool은 정확히 한 번 실행될 수 있을까의 action 저장·승인·idempotency key 패턴과 맞춰야 한다. 중요한 점은 데이터 흐름이다.

LangGraph update    → filtered Redis snapshot
LangGraph custom    → validated product event → Redis Stream
LangGraph result    → PostgreSQL 기준 결과
Exception           → attempt failure record → queue retry policy

LangGraph가 graph 실행을 진행시키지만, worker가 이번 실행 시도의 수명 주기와 외부 쓰기를 책임진다. SSE 연결은 여기에 참여하지 않는다. 이전 글 Agent가 일하면서 말하게 하기에서 만든 gateway는 Redis Stream을 cursor부터 읽을 뿐이다.

Checkpoint는 강력하지만 queue와 업무 데이터베이스를 대체하지 않는다

LangGraph checkpointer는 graph state snapshot을 thread 단위로 저장한다. 공식 영속성 문서는 이를 대화 연속성, human-in-the-loop, 시점 이동과 장애 허용에 사용하는 단기 thread 범위 메모리로 설명한다. 애플리케이션이 정의한 thread 간 데이터를 위한 store와도 구분한다. (LangGraph Persistence)

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver


async with AsyncPostgresSaver.from_conn_string(CHECKPOINT_DSN) as checkpointer:
    # 최초 schema 준비 시에는 await checkpointer.setup()을 별도로 실행한다.
    graph = builder.compile(checkpointer=checkpointer)

    config = {
        "configurable": {
            "thread_id": server_generated_checkpoint_key,
        }
    }

    result = await graph.ainvoke(graph_input, config=config)

Checkpoint가 답하는 질문은 이런 것이다.

• 이 graph는 어느 state에서 멈췄는가?
• 재개할 때 어느 node부터 이어갈 수 있는가?
• HITL interrupt 전후의 runtime state는 무엇인가?

송금처럼 부수 효과가 있는 Tool node는 checkpoint 직전에서 멈췄다가 재개하면 같은 node가 다시 실행될 수 있다. Checkpoint는 graph execution을 이어 주지만 외부 세계의 중복 실행을 막지는 않는다. Approval, action fingerprint, idempotency key는 Agent Tool은 정확히 한 번 실행될 수 있을까에서 다룬다.

반면 다음 질문은 별도 설계가 필요하다.

• 아직 실행되지 않은 run을 어느 worker에게 전달할 것인가?
• Worker가 죽은 뒤 누가 대기 중인 메시지를 회수하는가?
• 어느 tenant와 사용자가 이 run을 읽을 수 있는가?
• 사용자가 결제한 analysis가 한 번만 확정되었는가?
• 최종 보고서와 감사 기록의 보존 정책은 무엇인가?

Checkpoint       graph execution을 다시 잇기 위한 state
Queue            work delivery와 redelivery
Business DB      제품 기록과 기준 결과
Event stream     관찰 가능한 progress history

한 저장소가 물리적으로 여러 역할을 맡을 수는 있다. PostgreSQL에 checkpoint table과 run table을 함께 둘 수도 있다. 하지만 schema, 일관성 경계와 책임까지 같은 것은 아니다.

thread_id 역시 저장소 조회 key이지 권한 증명이 아니다. Client가 보낸 값을 그대로 신뢰하면 안 된다. API가 tenant와 사용자의 run 접근 권한을 확인한 뒤 서버 측 식별자를 graph config에 넣어야 한다. 이 구분은 다음 글에서 더 자세히 다룬다.

무엇을 graph 안에 넣고 무엇을 밖에 둘까

경계가 애매할 때는 “이 로직이 어떤 종류의 결정을 내리는가”를 묻는다.

관심사	기본 위치	이유
다음 분석 node 선택	LangGraph	현재 graph state에 따른 실행 전환
병렬 분기와 fan-in	LangGraph	node 의존성과 reducer 의미 체계
도구 결과를 다음 node에 전달	LangGraph state	다음 계산의 작업 데이터
Graph 중단·재개 지점	Checkpointer	runtime 이어가기
Run 생성 권한	API/애플리케이션	tenant와 제품 정책
대기 작업 전달	Queue	프로세스 수명을 넘는 책임
활성 실행 시도 heartbeat	Worker/애플리케이션	worker 실패와 회수 판단
브라우저 진행률 snapshot	Redis 투영값	빠른 읽기와 UI 복원
공개 SSE event schema	애플리케이션 계약	framework 독립성과 데이터 최소화
최종 보고서와 과금 status	업무 DB	제품의 기준 기록

항상 이 표와 같아야 한다는 뜻은 아니다. 예를 들어 관리형 Agent Server를 도입하면 queue, 영속성과 run API 일부를 제품이 맡을 수 있다. 그때도 표를 지우는 대신 각 행의 책임 주체를 관리형 서비스로 바꿔 적어야 한다.

LangGraph를 쓰지 말자는 글이 아니다

LangGraph는 다음 조건에서 특히 유용하다.

• 분기와 순환이 있는 stateful workflow를 명시적으로 보고 싶다.
• 병렬 node 갱신을 reducer 의미 체계로 합쳐야 한다.
• HITL 중단 뒤 같은 graph state에서 재개해야 한다.
• Node 갱신, token과 custom 진행률을 runtime에서 관찰해야 한다.
• Checkpoint와 실행 이력을 활용해야 한다.

반대로 세 단계의 고정 ETL처럼 일반 함수 호출만으로 충분한 작업에 graph를 넣으면 추상화 비용이 더 클 수 있다. 종료 상한이 없는 도구 loop 하나가 전부라면 직접 소유하고 실행 상한을 명시한 Python loop가 더 읽기 쉬울 수 있다.

선택 기준은 “Agent니까 LangGraph”가 아니다.

Control flow와 resume semantics가 복잡한가?
    ├─ Yes → Graph runtime의 가치가 커진다.
    └─ No  → Plain Python workflow가 더 단순할 수 있다.

이 관점은 에이전트는 꼭 그래프여야 할까에서 말한 “Agent 내부에서는 자유를, 시스템 경계에서는 통제를”라는 관점과도 이어진다. Graph는 아키텍처 전체의 동의어가 아니라, 명시적 실행 제어가 필요한 곳에 놓는 runtime이다.

API가 빠르게 변할수록 애플리케이션 경계가 중요하다

LangGraph의 streaming 표면은 계속 발전하고 있다. 현재 공식 문서는 새 애플리케이션에 LangGraph 1.2에서 도입된 event streaming API를 권장하면서도, updates, custom 등을 직접 소비하는 stream-mode API를 함께 문서화한다. 이 글은 runtime state와 애플리케이션 투영값의 경계를 가장 분명히 보여주기 위해 graph.astream(..., version="v2") stream-mode API를 사용했다.

여기서 얻을 교훈은 특정 version을 영원히 고정하라는 것이 아니다.

async def runtime_events(graph, graph_input, config):
    """Framework-specific adapter: keep this boundary narrow."""
    async for part in graph.astream(
        graph_input,
        config=config,
        stream_mode=["updates", "custom"],
        version="v2",
    ):
        yield part

Framework별 파싱을 adapter 한 곳에 모으고, 그 뒤에는 애플리케이션이 소유한 RunProgressed, ApprovalRequired, RunSucceeded 같은 계약을 사용한다. 그러면 새로운 streaming API로 옮겨도 Redis event schema, SSE client와 브라우저 구성 요소를 동시에 갈아엎지 않아도 된다.

구현 확인 목록

• LangGraph가 worker 안의 runtime이라는 위치가 명시되어 있는가?
• Graph state에 다음 node 실행에 필요하지 않은 UI 이력을 쌓고 있지 않은가?
• Reducer와 외부 데이터베이스 일관성을 혼동하지 않는가?
• updates와 전체 values의 차이를 이해하고 필요한 데이터만 소비하는가?
• 긴 node 내부 진행률은 검증된 custom event로 내보내는가?
• 원본 StreamPart나 비공개 graph 채널을 브라우저에 그대로 노출하지 않는가?
• LangGraph state, Redis UI state와 브라우저 state를 각각 정의했는가?
• Redis에는 큰 문서와 내부 prompt를 무조건 복제하지 않는가?
• Checkpoint, queue, 업무 DB와 event stream의 역할을 구분했는가?
• thread_id를 권한 token처럼 사용하지 않는가?
• Framework API를 좁은 adapter 뒤에 격리했는가?
• 최종 결과와 종료 event의 기록 순서 및 실패 정책을 정했는가?

결론: Runtime을 제자리에 놓으면 아키텍처가 보인다

LangGraph는 Agent의 실행을 표현하고 진행시키는 강력한 runtime이다. Node가 state의 일부를 갱신하고, reducer가 갱신을 합치며, 조건부 edge가 다음 경로를 고른다. updates는 state 전환을 관찰하게 하고, custom은 긴 node 내부의 진행 상황을 내보낸다. Checkpoint는 중단된 graph를 다시 잇는 기반이 된다.

하지만 그 기능이 queue 전달, worker 책임, 권한 확인, UI 투영값, SSE replay와 업무 기록을 자동으로 하나의 아키텍처로 만들지는 않는다.

핵심은 다음 두 문장이다.

LangGraph는 아키텍처가 아니라 아키텍처 안에 배치되는 Agent runtime이다.

LangGraph State, Redis UI State와 Browser UI State는 같을 필요가 없고, 대개 같아서는 안 된다.

이제 runtime의 위치를 정했다. 다음 문제는 이 실행을 누구의 대화와 연결하고, 동시에 들어온 여러 run을 어떻게 격리하느냐다.

다음 글 Conversation은 Run이 아니다: Session Isolation 설계에서는 Tenant → User → Session → Run의 식별 계층, thread_id의 정확한 역할, 권한 확인과 단일 활성 run 정책을 다룬다.