[LLM] LangGraph의 node 실행 방식

LangGraph 로 graph 를 만들다 보면 각 node가 어떻게 동작하는지, graph의 실행 방식에 대한 의문이 생긴다. node 를 sync 함수로 쓸 때와 async 함수로 쓸 때 무엇이 달라지는가. stream() 과 astream() 은 node 를 어떻게 다르게 실행하는가. sync node 를 astream() 으로 실행해도 되는가.

 

이번 글에서는 LangGraph 소스를 따라가며 node 가 실제로 어떻게 실행되는지 확인해보겠다. 기준 버전은 langgraph 1.2 이다.

요약

결론부터 먼저 보겠다.

 

	sync API (invoke/stream)	async API (ainvoke/astream)
sync node (def)	호출 스레드에서 직접 실행	워커 스레드에 위임 (executor)
async node (async def)	TypeError, 실행 불가	event loop 에서 직접 실행

 

이 표에서 가장 눈에 띄는 칸은 TypeError 다. 아래 예제로 바로 확인할 수 있다.

 

import asyncio
from typing import TypedDict
from langgraph.graph import END, START, StateGraph


class S(TypedDict):
    v: int


async def a_node(state: S) -> S:
    return {"v": state["v"] + 1}


g = StateGraph(S).add_node("a", a_node).add_edge(START, "a").add_edge("a", END).compile()

g.invoke({"v": 0})
# TypeError: No synchronous function provided to "a".
# Either initialize with a synchronous function or invoke via the async API (ainvoke, astream, etc.)

asyncio.run(g.ainvoke({"v": 0}))
# {'v': 1}

 

같은 graph 인데 sync API 인 invoke() 는 TypeError 를 내고 async API 인 ainvoke() 는 정상 실행된다. 이 표가 어떻게 만들어지는지 세 단계로 나눠서 보겠다: node 등록 시점, graph 실행 파이프라인, node 호출 지점이다.

1. node 등록 시점: RunnableCallable 의 두 슬롯

add_node() 로 넘긴 함수는 그대로 저장되지 않고 RunnableCallable 이라는 wrapper 로 감싸진다. 이 변환을 하는 coerce_to_runnable 을 보면 sync/async 분기는 이미 등록 시점에 끝나 있다.

 

# langgraph/_internal/_runnable.py, coerce_to_runnable (발췌)
if is_async_callable(thing):
    return RunnableCallable(None, thing, ...)
else:
    return RunnableCallable(
        thing,
        wraps(thing)(partial(run_in_executor, None, thing)),
        ...
    )

 

RunnableCallable(func, afunc) 는 sync 실행용 슬롯 func 와 async 실행용 슬롯 afunc 를 가진다.

• async node: func=None, afunc=원본. sync 실행용 슬롯이 비어 있다.
• sync node: func=원본, afunc 는 run_in_executor 로 감싼 래퍼가 미리 들어간다. "sync node 를 async 로 실행하면 스레드에 위임한다"는 동작은 실행 중의 판단이 아니라, 등록 시점에 생성된 래퍼 함수 그 자체이다.

실행 API 는 자기 슬롯만 꺼내 쓴다. sync 경로인 invoke 는 func 가 비어 있으면 에러를 낸다.

 

# RunnableCallable.invoke (발췌)
if self.func is None:
    raise TypeError(
        f'No synchronous function provided to "{self.name}".'
        "\nEither initialize with a synchronous function or invoke"
        " via the async API (ainvoke, astream, etc.)"
    )

 

글 처음의 최소 예제에서 본 에러가 바로 이것이다. 발생 시점도 이 구조에서 읽힌다. compile() 은 이 에러를 잡아주지 못한다. 실행이 해당 node 에 도달해 invoke 가 호출되는 순간에야 나는 런타임 에러이다. async node 를 하나라도 넣으면 호출부를 ainvoke()/astream() 으로 바꿔야 하는 이유가 코드 구조로 설명된다. 반대로 sync node 는 두 슬롯이 모두 차 있으므로 어느 API 로든 실행할 수 있다.

2. 실행 파이프라인: loop, runner, task

compile() 이 반환하는 graph 를 실행하면 다음 체인을 타고 내려간다.

 

sync API                              async API
graph.stream()                        graph.astream()
  → SyncPregelLoop (step 반복)          → AsyncPregelLoop (step 반복)
    → PregelRunner.tick(tasks)            → PregelRunner.atick(tasks)
      → run_with_retry(task)                → arun_with_retry(task)
        → task.proc.invoke(...)               → await task.proc.ainvoke(...)

 

Pregel 은 Google 의 대규모 graph 처리 모델의 이름으로, LangGraph 의 실행 엔진이 이 모델을 따르기 때문에 내부 클래스 이름에 그대로 쓰인다. task 는 node 실행 한 건을 뜻하고 task.proc 은 node 를 감싼 RunnableCallable 에 state 쓰기 같은 부가 동작을 붙인 실행 단위이다.

 

두 가지를 짚고 넘어가겠다.

 

첫째, 실행 단위는 node 나열이 아니라 super-step 이다. LangGraph 는 Pregel 모델을 따르는데, 여기서 super-step은 "현재 실행 가능한 모든 노드를 한 번에 실행하고 그 결과를 state에 반영하는 실행 단계"를 의미한다. 실행 루프는 이러한 super-step을 반복하며 동작한다. 매 step 마다 직전 step 의 state 변경 (edge) 으로 트리거된 node 들을 골라 task 를 만들고 그 step 의 task 를 전부 실행하고, 결과를 state 에 커밋한 뒤 다음 step 으로 넘어간다. START → A → END 같은 직선 graph 는 step 당 node 가 하나라 순차 실행처럼 보이지만 분기가 있으면 달라진다. 예를 들어 A 에서 B 와 C 로 갈라지는 graph 라면 B 와 C 는 같은 step 에 속하고 동시에 실행된다.

 

둘째, invoke/ainvoke 선택은 node 가 아니라 호출 API 가 정한다. 어느 API 로 graph 를 실행했는지가 체인 전체를 sync 계열 (tick 과 invoke) 또는 async 계열 (atick 과 ainvoke) 로 결정한다. node 의 sync/async 적응은 체인의 맨 끝, 1번에서 본 RunnableCallable 슬롯에서 일어난다.

 

# langgraph/pregel/_retry.py (발췌)
# sync 경로
return task.proc.invoke(task.input, config)
# async 경로
return await task.proc.ainvoke(task.input, config)

3. 병렬 실행과 fast path

같은 step 에 task 가 여러 개일 때의 병렬 수단이 API 에 따라 다르다.

• tick (sync): task 들을 thread pool 에 submit 해서 스레드로 동시에 실행한다.
• atick (async): task 들을 asyncio task 로 만들어 event loop 에서 동시에 실행한다.

tick 에는 최적화가 하나 있다. task 가 하나뿐이면 thread pool 에 보내지 않고 호출자 스레드에서 그대로 실행한다.

 

# langgraph/pregel/_runner.py, tick (발췌)
elif len(tasks) == 1 and timeout is None and get_waiter is None:
    t = tasks[0]
    ...
    run_with_retry(t, retry_policy, ...)

 

직선형 graph 를 stream() 으로 실행하면 매 step 이 이 fast path 를 타므로, node 는 graph.stream() 을 호출한 바로 그 스레드에서 실행된다. 스레드 홉이 아예 없는 것이다.

4. 조합별 실행 구조

이제 처음의 표를 실행 구조 관점으로 다시 확인해보자.

 

sync API + sync node: 순수 스레드 세계이다. event loop 가 없고 node 안의 블로킹 (ORM, time.sleep, sync LLM 호출) 은 호출 스레드가 그냥 기다린다. 스레드 모델에서는 정상 동작이다.

 

async API + sync node: event loop 가 run_in_executor 래퍼 (1번에서 등록 시점에 만들어진 그것) 를 통해 node 를 워커 스레드로 보낸다. 이때 contextvars 를 스레드로 복사하므로, executor 스레드 안에서도 get_stream_writer() (node 실행 도중 값을 graph 스트림으로 내보내는 기능) 같은 실행 컨텍스트 조회가 동작한다. node 가 스레드에서 블로킹하는 동안 event loop 는 자유롭다. 스트림 소비와 다른 요청 처리가 계속된다. node 중간에 writer 로 내보낸 값은 내부 큐를 거쳐 loop 쪽 소비자에게 실시간으로 전달된다. sync node 를 고치지 않고도 astream() 으로 토큰 스트리밍이 되는 이유가 이 구조이다.

 

sync API + async node: 1번에서 본 TypeError 이다. sync 실행 경로에는 coroutine 을 실행할 event loop 가 없고 LangGraph 는 조용히 우회하는 대신 에러를 낸다.

 

async API + async node: node 가 event loop 의 task 로 직접 실행된다. 스레드 홉이 없고 같은 step 의 병렬 node 들은 동시 task 가 되어 I/O 대기가 실제로 겹쳐진다. 대신 규율이 하나 필요하다: node 안의 모든 호출이 non-blocking 이어야 한다. async node 안에서 sync ORM 이나 sync LLM 클라이언트를 부르면 그 블로킹이 스레드가 아니라 event loop 전체를 멈춘다. 같은 프로세스의 모든 요청이 함께 정지하는 것이다. async node 안의 sync 코드는 sync_to_async 등으로 다시 스레드에 보내야 한다. async API 와 sync node 조합에서 LangGraph 가 자동으로 해주던 위임을, 이 조합에서는 개발자가 호출 단위로 직접 하는 셈이다.

정리

• sync/async 분기는 add_node() 등록 시점에 RunnableCallable 의 두 슬롯으로 만들어지고 실행 API 는 자기 슬롯을 꺼내 쓸 뿐이다.
• 실행은 super-step 단위이고 invoke/ainvoke 선택은 호출 API (stream() vs astream()) 를 따른다.
• sync node 는 양쪽 API 에서 모두 동작한다. async API 에서는 등록 시점에 만들어진 executor 래퍼가 스레드 위임을 해준다.
• async node 는 async API 를 강제하며 node 내부의 블로킹 호출을 스스로 관리해야 한다.

실무 관점의 함의는 명확하다: graph 를 async 로 전환할 때는 호출부 (astream()) 먼저, node 는 그 다음이다. 호출부만 바꿔도 sync node 들은 executor 위임으로 정상 동작하고 이후 병렬 I/O 이득이 필요한 node 부터 하나씩 async 화하면 된다. 반대 순서 (node 먼저 async 화) 는 TypeError 때문에 성립하지 않는다.